[misc] adding option transactionReplaySize to control redo cache size

Author: diego Dupin

src/main/java/org/mariadb/jdbc/Configuration.java

@@ -127,6 +127,7 @@ public class Configuration {
  private int retriesAllDown = 120;
  private String galeraAllowedState = null;
  private boolean transactionReplay = false;
+ private int transactionReplaySize = 64;
 
  // Pool options
  private boolean pool = false;
@@ -202,6 +203,7 @@ private Configuration(
  int retriesAllDown,
  String galeraAllowedState,
  boolean transactionReplay,
+ int transactionReplaySize,
  boolean pool,
  String poolName,
  int maxPoolSize,
@@ -267,6 +269,7 @@ private Configuration(
  this.retriesAllDown = retriesAllDown;
  this.galeraAllowedState = galeraAllowedState;
  this.transactionReplay = transactionReplay;
+ this.transactionReplaySize = transactionReplaySize;
  this.pool = pool;
  this.poolName = poolName;
  this.maxPoolSize = maxPoolSize;
@@ -343,6 +346,7 @@ private Configuration(
  Boolean useReadAheadInput,
  Boolean cachePrepStmts,
  Boolean transactionReplay,
+ Integer transactionReplaySize,
  String geometryDefaultType,
  String restrictedAuth,
  Properties nonMappedOptions)
@@ -421,6 +425,7 @@ private Configuration(
  if (useReadAheadInput != null) this.useReadAheadInput = useReadAheadInput;
  if (cachePrepStmts != null) this.cachePrepStmts = cachePrepStmts;
  if (transactionReplay != null) this.transactionReplay = transactionReplay;
+ if (transactionReplaySize != null) this.transactionReplaySize = transactionReplaySize;
  if (geometryDefaultType != null) this.geometryDefaultType = geometryDefaultType;
  if (restrictedAuth != null) this.restrictedAuth = restrictedAuth;
  if (serverSslCert != null) this.serverSslCert = serverSslCert;
@@ -720,6 +725,7 @@ public Configuration clone(String username, String password) {
  this.retriesAllDown,
  this.galeraAllowedState,
  this.transactionReplay,
+ this.transactionReplaySize,
  this.pool,
  this.poolName,
  this.maxPoolSize,
@@ -993,6 +999,10 @@ public boolean transactionReplay() {
  return transactionReplay;
  }
 
+ public int transactionReplaySize() {
+ return transactionReplaySize;
+ }
+
  public String geometryDefaultType() {
  return geometryDefaultType;
  }
@@ -1235,6 +1245,7 @@ public static final class Builder implements Cloneable {
  private Integer retriesAllDown;
  private String galeraAllowedState;
  private Boolean transactionReplay;
+ private Integer transactionReplaySize;
 
  // Pool options
  private Boolean pool;
@@ -1714,6 +1725,11 @@ public Builder transactionReplay(Boolean transactionReplay) {
  return this;
  }
 
+ public Builder transactionReplaySize(Integer transactionReplaySize) {
+ this.transactionReplaySize = transactionReplaySize;
+ return this;
+ }
+
  public Configuration build() throws SQLException {
  Configuration conf =
  new Configuration(
@@ -1779,6 +1795,7 @@ public Configuration build() throws SQLException {
  this.useReadAheadInput,
  this.cachePrepStmts,
  this.transactionReplay,
+ this.transactionReplaySize,
  this.geometryDefaultType,
  this.restrictedAuth,
  this._nonMappedOptions);

src/main/java/org/mariadb/jdbc/ServerPreparedStatement.java

@@ -50,21 +50,13 @@ public ServerPreparedStatement(
  resultSetType,
  resultSetConcurrency,
  defaultFetchSize);
- if (!PREPARABLE_STATEMENT_PATTERN.matcher(sql).find()) {
- prepareIfNotAlready(sql);
+ prepareResult = con.getContext().getPrepareCache().get(sql, this);
+ if (prepareResult == null && !PREPARABLE_STATEMENT_PATTERN.matcher(sql).find()) {
+ con.getClient().execute(new PreparePacket(sql), this);
  }
  parameters = new ParameterList();
  }
 
- private void prepareIfNotAlready(String cmd) throws SQLException {
- if (prepareResult == null) {
- prepareResult = con.getContext().getPrepareCache().get(cmd, this);
- if (prepareResult == null) {
- con.getClient().execute(new PreparePacket(cmd), this);
- }
- }
- }
-
  protected void executeInternal() throws SQLException {
  checkNotClosed();
  validParameters();
@@ -119,7 +111,12 @@ private void executePipeline(String cmd) throws SQLException {
 
  private void executeStandard(String cmd) throws SQLException {
  // send COM_STMT_PREPARE
- prepareIfNotAlready(cmd);
+ if (prepareResult == null) {
+ prepareResult = con.getContext().getPrepareCache().get(cmd, this);
+ if (prepareResult == null) {
+ con.getClient().execute(new PreparePacket(cmd), this);
+ }
+ }
 
  // send COM_STMT_EXECUTE
  ExecutePacket execute = new ExecutePacket(prepareResult, parameters, cmd, this);

src/main/java/org/mariadb/jdbc/client/context/RedoContext.java

@@ -24,7 +24,7 @@ public RedoContext(
  ExceptionFactory exceptionFactory,
  PrepareCache prepareCache) {
  super(handshake, clientCapabilities, conf, exceptionFactory, prepareCache);
- this.transactionSaver = new TransactionSaver();
+ this.transactionSaver = new TransactionSaver(conf.transactionReplaySize());
  }
 
  public void setServerStatus(int serverStatus) {

src/main/java/org/mariadb/jdbc/client/impl/MultiPrimaryClient.java

@@ -134,7 +134,7 @@ protected Client connectHost(boolean readOnly, boolean failFast) throws SQLExcep
  throw lastSqle;
  }
 
- protected void reConnect() throws SQLException {
+ protected Client reConnect() throws SQLException {
 
  denyList.putIfAbsent(
  currentClient.getHostAddress(), System.currentTimeMillis() + deniedListTimeout);
@@ -146,35 +146,54 @@ protected void reConnect() throws SQLException {
 
  currentClient = connectHost(false, false);
  syncNewState(oldClient);
+ return oldClient;
 
- if (conf.transactionReplay()) {
- executeTransactionReplay(oldClient);
- } else if ((oldClient.getContext().getServerStatus() & ServerStatus.IN_TRANSACTION) > 0) {
- // transaction is lost, but connection is now up again.
+ } catch (SQLNonTransientConnectionException sqle) {
+ currentClient = null;
+ closed = true;
+ throw sqle;
+ }
+ }
+
+ protected void replayIfPossible(Client oldClient) throws SQLException {
+ // oldClient is only valued if this occurs on master.
+ if (oldClient != null) {
+ if ((oldClient.getContext().getServerStatus() & ServerStatus.IN_TRANSACTION) > 0) {
+ if (conf.transactionReplay()) {
+ executeTransactionReplay(oldClient);
+ } else {
+ // transaction is lost, but connection is now up again.
+ // changing exception to SQLTransientConnectionException
+ throw new SQLTransientConnectionException(
+ String.format(
+ "Driver has reconnect connection after a communications link failure with %s. In progress transaction was lost",
+ oldClient.getHostAddress()),
+ "25S03");
+ }
+ } else {
+ // no transaction, but connection is now up again.
  // changing exception to SQLTransientConnectionException
  throw new SQLTransientConnectionException(
  String.format(
- "Driver has reconnect connection after a "
- + "communications "
- + "link "
- + "failure with %s. In progress transaction was lost",
+ "Driver has reconnect connection after a communications link failure with %s",
  oldClient.getHostAddress()),
  "25S03");
  }
-
- } catch (SQLNonTransientConnectionException sqle) {
- currentClient = null;
- closed = true;
- throw sqle;
  }
  }
 
  protected void executeTransactionReplay(Client oldCli) throws SQLException {
  // transaction replay
- if ((oldCli.getContext().getServerStatus() & ServerStatus.IN_TRANSACTION) > 0) {
- RedoContext ctx = (RedoContext) oldCli.getContext();
- ((ReplayClient) currentClient).transactionReplay(ctx.getTransactionSaver());
+ RedoContext ctx = (RedoContext) oldCli.getContext();
+ if (ctx.getTransactionSaver().isDirty()) {
+ ctx.getTransactionSaver().clear();
+ throw new SQLTransientConnectionException(
+ String.format(
+ "Driver has reconnect connection after a communications link failure with %s. In progress transaction was too big to be replayed, and was lost",
+ oldCli.getHostAddress()),
+ "25S03");
  }
+ ((ReplayClient) currentClient).transactionReplay(ctx.getTransactionSaver());
  }
 
  public void syncNewState(Client oldCli) throws SQLException {
@@ -271,18 +290,18 @@ public List<Completion> execute(
  closeOnCompletion);
  } catch (SQLNonTransientConnectionException e) {
  HostAddress hostAddress = currentClient.getHostAddress();
- reConnect();
+ Client oldClient = reConnect();
 
  if (message instanceof QueryPacket && ((QueryPacket) message).isCommit()) {
  throw new SQLTransientConnectionException(
  String.format(
- "Driver has reconnect connection after a "
- + "communications "
- + "failure with %s during a COMMIT statement",
+ "Driver has reconnect connection after a communications failure with %s during a COMMIT statement",
  hostAddress),
  "25S03");
  }
 
+ replayIfPossible(oldClient);
+
  if (message instanceof RedoableWithPrepareClientMessage) {
  ((RedoableWithPrepareClientMessage) message).rePrepare(currentClient);
  }
@@ -323,7 +342,8 @@ public List<Completion> executePipeline(
  } catch (SQLException e) {
  if (e instanceof SQLNonTransientConnectionException
  || (e.getCause() != null && e.getCause() instanceof SQLNonTransientConnectionException)) {
- reConnect();
+ Client oldClient = reConnect();
+ replayIfPossible(oldClient);
  Arrays.stream(messages)
  .filter(RedoableWithPrepareClientMessage.class::isInstance)
  .map(RedoableWithPrepareClientMessage.class::cast)
@@ -365,8 +385,13 @@ public void readStreamingResults(
  currentClient.readStreamingResults(
  completions, fetchSize, maxRows, resultSetConcurrency, resultSetType, closeOnCompletion);
  } catch (SQLNonTransientConnectionException e) {
- reConnect();
- throw getExceptionFactory().create("Socket error during result streaming", "HY000");
+ try {
+ reConnect();
+ } catch (SQLException e2) {
+ throw getExceptionFactory()
+ .create("Socket error during result streaming", e2.getSQLState(), e2);
+ }
+ throw getExceptionFactory().create("Socket error during result streaming", "HY000", e);
  }
  }
 

src/main/java/org/mariadb/jdbc/client/impl/MultiPrimaryReplicaClient.java

@@ -6,7 +6,6 @@
 
 import java.sql.SQLException;
 import java.sql.SQLNonTransientConnectionException;
-import java.sql.SQLTransientConnectionException;
 import java.util.List;
 import java.util.concurrent.Executor;
 import java.util.concurrent.locks.ReentrantLock;
@@ -19,7 +18,6 @@
 import org.mariadb.jdbc.export.ExceptionFactory;
 import org.mariadb.jdbc.export.Prepare;
 import org.mariadb.jdbc.message.ClientMessage;
-import org.mariadb.jdbc.util.constants.ServerStatus;
 import org.mariadb.jdbc.util.log.Logger;
 import org.mariadb.jdbc.util.log.Loggers;
 
@@ -91,7 +89,7 @@ private void reconnectIfNeeded() {
  * @throws SQLException if exception
  */
  @Override
- protected void reConnect() throws SQLException {
+ protected Client reConnect() throws SQLException {
  denyList.putIfAbsent(
  currentClient.getHostAddress(), System.currentTimeMillis() + deniedListTimeout);
  logger.info("Connection error on {}", currentClient.getHostAddress());
@@ -155,22 +153,7 @@ protected void reConnect() throws SQLException {
 
  // if reconnect succeed on replica / use master, no problem, continuing without interruption
  // if reconnect primary, then replay transaction / throw exception if was in transaction.
- if (!requestReadOnly) {
- if (conf.transactionReplay()) {
- executeTransactionReplay(oldClient);
- } else if ((oldClient.getContext().getServerStatus() & ServerStatus.IN_TRANSACTION) > 0) {
- // transaction is lost, but connection is now up again.
- // changing exception to SQLTransientConnectionException
- throw new SQLTransientConnectionException(
- String.format(
- "Driver has reconnect connection after a "
- + "communications "
- + "link "
- + "failure with %s. In progress transaction was lost",
- oldClient.getHostAddress()),
- "25S03");
- }
- }
+ return requestReadOnly ? null : oldClient;
 
  } catch (SQLNonTransientConnectionException sqle) {
  currentClient = null;

src/main/java/org/mariadb/jdbc/client/impl/ReplayClient.java

@@ -103,11 +103,12 @@ public List<Completion> execute(
  }
 
  public void transactionReplay(TransactionSaver transactionSaver) throws SQLException {
- List<RedoableClientMessage> buffers = transactionSaver.getBuffers();
+ RedoableClientMessage[] buffers = transactionSaver.getBuffers();
  try {
  // replay all but last
  Prepare prepare;
- for (RedoableClientMessage querySaver : buffers) {
+ for (int i = 0; i < transactionSaver.getIdx(); i++) {
+ RedoableClientMessage querySaver = buffers[i];
  int responseNo;
  if (querySaver instanceof RedoableWithPrepareClientMessage) {
  // command is a prepare statement query

src/main/java/org/mariadb/jdbc/client/impl/TransactionSaver.java

@@ -4,22 +4,41 @@
 
 package org.mariadb.jdbc.client.impl;
 
-import java.util.ArrayList;
-import java.util.List;
+import java.util.Arrays;
 import org.mariadb.jdbc.message.client.RedoableClientMessage;
 
 public class TransactionSaver {
- private final List<RedoableClientMessage> buffers = new ArrayList<>();
+ private final RedoableClientMessage[] buffers;
+ private int idx = 0;
+ private boolean dirty = false;
+
+ public TransactionSaver(int transactionReplaySize) {
+ buffers = new RedoableClientMessage[transactionReplaySize];
+ }
 
  public void add(RedoableClientMessage clientMessage) {
- buffers.add(clientMessage);
+ if (idx < buffers.length) {
+ buffers[idx++] = clientMessage;
+ } else {
+ dirty = true;
+ }
  }
 
  public void clear() {
- buffers.clear();
+ Arrays.fill(buffers, null);
+ dirty = false;
+ idx = 0;
+ }
+
+ public int getIdx() {
+ return idx;
+ }
+
+ public boolean isDirty() {
+ return dirty;
  }
 
- public List<RedoableClientMessage> getBuffers() {
+ public RedoableClientMessage[] getBuffers() {
  return buffers;
  }
 }

src/main/resources/driver.properties

@@ -49,6 +49,7 @@ useReadAheadInput=use a buffered inputSteam that read socket available data. Thi
 cachePrepStmts=enable/disable prepare Statement cache. When enable, PreparedStatement.close won't close prepare immediately, keeping a pool of most used prepared results. Default true.
 timezone=Driver set connection timezone to client timezone. Value default to client default time-zone ID, but can be set using this option. If you are sure that server use the same timezone, setting connection timezone can be disabled using value 'disable'.
 transactionReplay=When having a failover, can current transaction beeing re-executed, having a completly transparent failover. All commands must be idempotent. Default false.
+transactionReplaySize=replay cache buffer maximum size. If a transaction has more command that this size and a failover occurs, transaction will then not be replayed, just throwing an exception error. (Integer) default 64.
 allowLocalInfile=Indicate if LOAD DATA LOCAL INFILE commands are permitted. This will disable all pipelining implementation. Default false.
 geometryDefaultType=Indicate what default Object type Geometry a resultset.getObject must return. null or empty is WKB byte array. 'default' will return org.mariadb.mariadb.jdbc.type Object. Default null
 keyStore=File path of the keyStore file that contain client private key store and associate certificates (similar to java System property "javax.net.ssl.keyStore", but ensure that only the private key's entries are used)