ARTEMIS-370: Make JDBC Journal Sync period configurable

artemis-core-client/src/main/java/org/apache/activemq/artemis/api/config/ActiveMQDefaultConfiguration.java

@@ -447,6 +447,8 @@ public final class ActiveMQDefaultConfiguration {
 
    private static final long DEFAULT_JDBC_LOCK_EXPIRATION_MILLIS = TimeUnit.SECONDS.toMillis(20);
 
+   private static final long DEFAULT_JDBC_JOURNAL_SYNC_PERIOD_MILLIS = 5;
+
    private static final long DEFAULT_JDBC_LOCK_ACQUISITION_TIMEOUT_MILLIS = -1;
 
    // Default period to wait between connection TTL checks
@@ -1242,6 +1244,10 @@ public final class ActiveMQDefaultConfiguration {
       return DEFAULT_JDBC_LOCK_RENEW_PERIOD_MILLIS;
    }
 
+   public static long getDefaultJdbcJournalSyncPeriodMillis() {
+      return DEFAULT_JDBC_JOURNAL_SYNC_PERIOD_MILLIS;
+   }
+
    public static long getDefaultJdbcLockExpirationMillis() {
       return DEFAULT_JDBC_LOCK_EXPIRATION_MILLIS;
    }

artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/journal/JDBCJournalImpl.java

@@ -56,7 +56,9 @@ public class JDBCJournalImpl extends AbstractJDBCDriver implements Journal {
    private static final Logger logger = Logger.getLogger(JDBCJournalImpl.class);
 
    // Sync Delay in ms
-   private static final int SYNC_DELAY = 5;
+   //private static final int SYNC_DELAY = 5;
+
+   private long syncDelay;
 
    private static int USER_VERSION = 1;
 
@@ -95,12 +97,14 @@ public class JDBCJournalImpl extends AbstractJDBCDriver implements Journal {
                           String tableName,
                           ScheduledExecutorService scheduledExecutorService,
                           Executor completeExecutor,
-                          IOCriticalErrorListener criticalIOErrorListener) {
+                          IOCriticalErrorListener criticalIOErrorListener,
+                          long syncDelay) {
       super(dataSource, provider);
       records = new ArrayList<>();
       this.scheduledExecutorService = scheduledExecutorService;
       this.completeExecutor = completeExecutor;
       this.criticalIOErrorListener = criticalIOErrorListener;
+      this.syncDelay = syncDelay;
    }
 
    public JDBCJournalImpl(String jdbcUrl,
@@ -108,18 +112,20 @@ public class JDBCJournalImpl extends AbstractJDBCDriver implements Journal {
                           SQLProvider sqlProvider,
                           ScheduledExecutorService scheduledExecutorService,
                           Executor completeExecutor,
-                          IOCriticalErrorListener criticalIOErrorListener) {
+                          IOCriticalErrorListener criticalIOErrorListener,
+                          long syncDelay) {
       super(sqlProvider, jdbcUrl, jdbcDriverClass);
       records = new ArrayList<>();
       this.scheduledExecutorService = scheduledExecutorService;
       this.completeExecutor = completeExecutor;
       this.criticalIOErrorListener = criticalIOErrorListener;
+      this.syncDelay = syncDelay;
    }
 
    @Override
    public void start() throws SQLException {
       super.start();
-      syncTimer = new JDBCJournalSync(scheduledExecutorService, completeExecutor, SYNC_DELAY, TimeUnit.MILLISECONDS, this);
+      syncTimer = new JDBCJournalSync(scheduledExecutorService, completeExecutor, syncDelay, TimeUnit.MILLISECONDS, this);
       started = true;
    }
 

artemis-server/src/main/java/org/apache/activemq/artemis/core/config/storage/DatabaseStorageConfiguration.java

@@ -50,6 +50,8 @@ public class DatabaseStorageConfiguration implements StoreConfiguration {
 
    private long jdbcLockAcquisitionTimeoutMillis = ActiveMQDefaultConfiguration.getDefaultJdbcLockAcquisitionTimeoutMillis();
 
+   private long jdbcJournalSyncPeriodMillis = ActiveMQDefaultConfiguration.getDefaultJdbcJournalSyncPeriodMillis();
+
    @Override
    public StoreType getStoreType() {
       return StoreType.DATABASE;
@@ -175,4 +177,12 @@ public class DatabaseStorageConfiguration implements StoreConfiguration {
    public void setJdbcLockAcquisitionTimeoutMillis(long jdbcLockAcquisitionTimeoutMillis) {
       this.jdbcLockAcquisitionTimeoutMillis = jdbcLockAcquisitionTimeoutMillis;
    }
+
+   public long getJdbcJournalSyncPeriodMillis() {
+      return jdbcJournalSyncPeriodMillis;
+   }
+
+   public void setJdbcJournalSyncPeriodMillis(long jdbcJournalSyncPeriodMillis) {
+      this.jdbcJournalSyncPeriodMillis = jdbcJournalSyncPeriodMillis;
+   }
 }

artemis-server/src/main/java/org/apache/activemq/artemis/core/deployers/impl/FileConfigurationParser.java

@@ -1472,6 +1472,7 @@ public final class FileConfigurationParser extends XMLConfigurationUtil {
       conf.setJdbcNetworkTimeout(getInteger(storeNode, "jdbc-network-timeout", conf.getJdbcNetworkTimeout(), Validators.NO_CHECK));
       conf.setJdbcLockRenewPeriodMillis(getLong(storeNode, "jdbc-lock-renew-period", conf.getJdbcLockRenewPeriodMillis(), Validators.NO_CHECK));
       conf.setJdbcLockExpirationMillis(getLong(storeNode, "jdbc-lock-expiration", conf.getJdbcLockExpirationMillis(), Validators.NO_CHECK));
+      conf.setJdbcJournalSyncPeriodMillis(getLong(storeNode, "jdbc-journal-sync-period", conf.getJdbcJournalSyncPeriodMillis(), Validators.NO_CHECK));
       return conf;
    }
 

artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/JDBCJournalStorageManager.java

@@ -67,13 +67,13 @@ public class JDBCJournalStorageManager extends JournalStorageManager {
             if (sqlProviderFactory == null) {
                sqlProviderFactory = new PropertySQLProvider.Factory(dbConf.getDataSource());
             }
-            bindingsJournal = new JDBCJournalImpl(dbConf.getDataSource(), sqlProviderFactory.create(dbConf.getBindingsTableName(), SQLProvider.DatabaseStoreType.BINDINGS_JOURNAL), dbConf.getBindingsTableName(), scheduledExecutorService, executorFactory.getExecutor(), criticalErrorListener);
+            bindingsJournal = new JDBCJournalImpl(dbConf.getDataSource(), sqlProviderFactory.create(dbConf.getBindingsTableName(), SQLProvider.DatabaseStoreType.BINDINGS_JOURNAL), dbConf.getBindingsTableName(), scheduledExecutorService, executorFactory.getExecutor(), criticalErrorListener,dbConf.getJdbcJournalSyncPeriodMillis());
-            messageJournal = new JDBCJournalImpl(dbConf.getDataSource(), sqlProviderFactory.create(dbConf.getMessageTableName(), SQLProvider.DatabaseStoreType.MESSAGE_JOURNAL), dbConf.getMessageTableName(), scheduledExecutorService, executorFactory.getExecutor(), criticalErrorListener);
+            messageJournal = new JDBCJournalImpl(dbConf.getDataSource(), sqlProviderFactory.create(dbConf.getMessageTableName(), SQLProvider.DatabaseStoreType.MESSAGE_JOURNAL), dbConf.getMessageTableName(), scheduledExecutorService, executorFactory.getExecutor(), criticalErrorListener, dbConf.getJdbcJournalSyncPeriodMillis());
             largeMessagesFactory = new JDBCSequentialFileFactory(dbConf.getDataSource(), sqlProviderFactory.create(dbConf.getLargeMessageTableName(), SQLProvider.DatabaseStoreType.LARGE_MESSAGE), executorFactory.getExecutor(), criticalErrorListener);
          } else {
             String driverClassName = dbConf.getJdbcDriverClassName();
-            bindingsJournal = new JDBCJournalImpl(dbConf.getJdbcConnectionUrl(), driverClassName, JDBCUtils.getSQLProvider(driverClassName, dbConf.getBindingsTableName(), SQLProvider.DatabaseStoreType.BINDINGS_JOURNAL), scheduledExecutorService, executorFactory.getExecutor(), criticalErrorListener);
+            bindingsJournal = new JDBCJournalImpl(dbConf.getJdbcConnectionUrl(), driverClassName, JDBCUtils.getSQLProvider(driverClassName, dbConf.getBindingsTableName(), SQLProvider.DatabaseStoreType.BINDINGS_JOURNAL), scheduledExecutorService, executorFactory.getExecutor(), criticalErrorListener, dbConf.getJdbcJournalSyncPeriodMillis());
-            messageJournal = new JDBCJournalImpl(dbConf.getJdbcConnectionUrl(), driverClassName, JDBCUtils.getSQLProvider(driverClassName, dbConf.getMessageTableName(), SQLProvider.DatabaseStoreType.MESSAGE_JOURNAL), scheduledExecutorService, executorFactory.getExecutor(), criticalErrorListener);
+            messageJournal = new JDBCJournalImpl(dbConf.getJdbcConnectionUrl(), driverClassName, JDBCUtils.getSQLProvider(driverClassName, dbConf.getMessageTableName(), SQLProvider.DatabaseStoreType.MESSAGE_JOURNAL), scheduledExecutorService, executorFactory.getExecutor(), criticalErrorListener, dbConf.getJdbcJournalSyncPeriodMillis());
             largeMessagesFactory = new JDBCSequentialFileFactory(dbConf.getJdbcConnectionUrl(), driverClassName, JDBCUtils.getSQLProvider(driverClassName, dbConf.getLargeMessageTableName(), SQLProvider.DatabaseStoreType.LARGE_MESSAGE), executorFactory.getExecutor(), criticalErrorListener);
          }
          final int networkTimeout = dbConf.getJdbcNetworkTimeout();

artemis-server/src/main/resources/schema/artemis-configuration.xsd

@@ -1956,6 +1956,13 @@
                </xsd:documentation>
             </xsd:annotation>
          </xsd:element>
+         <xsd:element name="jdbc-journal-sync-period" type="xsd:string" minOccurs="0" maxOccurs="1">
+            <xsd:annotation>
+               <xsd:documentation>
+                  The JDBC jouranl sync period in milliseconds.
+               </xsd:documentation>
+            </xsd:annotation>
+         </xsd:element>
       </xsd:all>
       <xsd:attributeGroup ref="xml:specialAttrs"/>
    </xsd:complexType>

tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jdbc/store/journal/JDBCJournalTest.java

@@ -86,7 +86,7 @@ public class JDBCJournalTest extends ActiveMQTestBase {
          public void onIOException(Throwable code, String message, SequentialFile file) {
 
          }
-      });
+      },5);
       journal.start();
    }