[AMQ-7473] Add stopOnError configuration to stop the broker when locker has an exception

activemq-broker/src/main/java/org/apache/activemq/broker/AbstractLocker.java

@@ -57,4 +57,5 @@ public abstract class AbstractLocker extends ServiceSupport implements Locker {
     public void setLockable(LockableServiceSupport lockableServiceSupport) {
         this.lockable = lockableServiceSupport;
     }
+
 }

activemq-broker/src/main/java/org/apache/activemq/broker/Lockable.java

@@ -31,6 +31,13 @@ public interface Lockable {
      */
     public void setUseLock(boolean useLock);
 
+    /**
+     * Stop the broker if the locker get an exception while processing lock.
+     *
+     * @param stopOnError
+     */
+    public void setStopOnError(boolean stopOnError);
+
     /**
      * Create a default locker
      *

activemq-broker/src/main/java/org/apache/activemq/broker/LockableServiceSupport.java

@@ -36,6 +36,7 @@ public abstract class LockableServiceSupport extends ServiceSupport implements L
 
     private static final Logger LOG = LoggerFactory.getLogger(LockableServiceSupport.class);
     boolean useLock = true;
+    boolean stopOnError = false;
     Locker locker;
     long lockKeepAlivePeriod = 0;
     private ScheduledFuture<?> keepAliveTicket;
@@ -58,6 +59,15 @@ public abstract class LockableServiceSupport extends ServiceSupport implements L
         return this.useLock;
     }
 
+    @Override
+    public void setStopOnError(boolean stopOnError) {
+        this.stopOnError = stopOnError;
+    }
+
+    public boolean isStopOnError() {
+        return this.stopOnError;
+    }
+
     @Override
     public void setLocker(Locker locker) throws IOException {
         this.locker = locker;
@@ -129,8 +139,14 @@ public abstract class LockableServiceSupport extends ServiceSupport implements L
                 }
             }
         } catch (SuppressReplyException e) {
+            if (stopOnError) {
+                stop = true;
+            }
             LOG.warn("locker keepAlive resulted in", e);
         } catch (IOException e) {
+            if (stopOnError) {
+                stop = true;
+            }
             LOG.warn("locker keepAlive resulted in", e);
         }
         if (stop) {

activemq-jdbc-store/src/main/java/org/apache/activemq/store/jdbc/JDBCPersistenceAdapter.java

@@ -94,6 +94,8 @@ public class JDBCPersistenceAdapter extends DataSourceServiceSupport implements
     private int transactionIsolation;
     private File directory;
     private boolean changeAutoCommitAllowed = true;
+    private int queryTimeout = -1;
+    private int networkTimeout = -1;
 
     protected int maxProducersToAudit=1024;
     protected int maxAuditDepth=1000;
@@ -513,7 +515,7 @@ public class JDBCPersistenceAdapter extends DataSourceServiceSupport implements
     }
 
     public TransactionContext getTransactionContext() throws IOException {
-        TransactionContext answer = new TransactionContext(this);
+        TransactionContext answer = new TransactionContext(this, networkTimeout, queryTimeout);
         if (transactionIsolation > 0) {
             answer.setTransactionIsolation(transactionIsolation);
         }
@@ -564,6 +566,32 @@ public class JDBCPersistenceAdapter extends DataSourceServiceSupport implements
         this.changeAutoCommitAllowed = changeAutoCommitAllowed;
     }
 
+    public int getNetworkTimeout() {
+        return networkTimeout;
+    }
+
+    /**
+     * Define the JDBC connection network timeout.
+     *
+     * @param networkTimeout the connection network timeout (in milliseconds).
+     */
+    public void setNetworkTimeout(int networkTimeout) {
+        this.networkTimeout = networkTimeout;
+    }
+
+    public int getQueryTimeout() {
+        return queryTimeout;
+    }
+
+    /**
+     * Define the JDBC statement query timeout.
+     *
+     * @param queryTimeout the statement query timeout (in seconds).
+     */
+    public void setQueryTimeout(int queryTimeout) {
+        this.queryTimeout = queryTimeout;
+    }
+
     @Override
     public void deleteAllMessages() throws IOException {
         TransactionContext c = getTransactionContext();

activemq-jdbc-store/src/main/java/org/apache/activemq/store/jdbc/TransactionContext.java

@@ -22,6 +22,7 @@ import java.util.LinkedList;
 import java.util.Map;
 import java.util.Properties;
 import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 
@@ -49,10 +50,14 @@ public class TransactionContext {
     private int transactionIsolation = Connection.TRANSACTION_READ_UNCOMMITTED;
     private LinkedList<Runnable> completions = new LinkedList<Runnable>();
     private ReentrantReadWriteLock exclusiveConnectionLock = new ReentrantReadWriteLock();
+    private int networkTimeout;
+    private int queryTimeout;
 
-    public TransactionContext(JDBCPersistenceAdapter persistenceAdapter) throws IOException {
+    public TransactionContext(JDBCPersistenceAdapter persistenceAdapter, int networkTimeout, int queryTimeout) throws IOException {
         this.persistenceAdapter = persistenceAdapter;
         this.dataSource = persistenceAdapter.getDataSource();
+        this.networkTimeout = networkTimeout;
+        this.queryTimeout = queryTimeout;
     }
 
     public Connection getExclusiveConnection() throws IOException {
@@ -68,6 +73,9 @@ public class TransactionContext {
             toLock.lock();
             try {
                 connection = dataSource.getConnection();
+                if (networkTimeout > 0) {
+                    connection.setNetworkTimeout(Executors.newSingleThreadExecutor(), networkTimeout);
+                }
                 if (persistenceAdapter.isChangeAutoCommitAllowed()) {
                     boolean autoCommit = !inTx;
                     if (connection.getAutoCommit() != autoCommit) {
@@ -300,17 +308,29 @@ public class TransactionContext {
         // simple delegate for the  rest of the impl..
         @Override
         public Statement createStatement() throws SQLException {
-            return delegate.createStatement();
+            Statement statement = delegate.createStatement();
+            if (queryTimeout > 0) {
+                statement.setQueryTimeout(queryTimeout);
+            }
+            return statement;
         }
 
         @Override
         public PreparedStatement prepareStatement(String sql) throws SQLException {
-            return delegate.prepareStatement(sql);
+            PreparedStatement statement = delegate.prepareStatement(sql);
+            if (queryTimeout > 0) {
+                statement.setQueryTimeout(queryTimeout);
+            }
+            return statement;
         }
 
         @Override
         public CallableStatement prepareCall(String sql) throws SQLException {
-            return delegate.prepareCall(sql);
+            CallableStatement statement = delegate.prepareCall(sql);
+            if (queryTimeout > 0) {
+                statement.setQueryTimeout(queryTimeout);
+            }
+            return statement;
         }
 
         @Override
@@ -390,17 +410,29 @@ public class TransactionContext {
 
         @Override
         public Statement createStatement(int resultSetType, int resultSetConcurrency) throws SQLException {
-            return delegate.createStatement(resultSetType, resultSetConcurrency);
+            Statement statement = delegate.createStatement(resultSetType, resultSetConcurrency);
+            if (queryTimeout > 0) {
+                statement.setQueryTimeout(queryTimeout);
+            }
+            return statement;
         }
 
         @Override
         public PreparedStatement prepareStatement(String sql, int resultSetType, int resultSetConcurrency) throws SQLException {
-            return delegate.prepareStatement(sql, resultSetType, resultSetConcurrency);
+            PreparedStatement statement = delegate.prepareStatement(sql, resultSetType, resultSetConcurrency);
+            if (queryTimeout > 0) {
+                statement.setQueryTimeout(queryTimeout);
+            }
+            return statement;
         }
 
         @Override
         public CallableStatement prepareCall(String sql, int resultSetType, int resultSetConcurrency) throws SQLException {
-            return delegate.prepareCall(sql, resultSetType, resultSetConcurrency);
+            CallableStatement statement = delegate.prepareCall(sql, resultSetType, resultSetConcurrency);
+            if (queryTimeout > 0) {
+                statement.setQueryTimeout(queryTimeout);
+            }
+            return statement;
         }
 
         @Override
@@ -445,32 +477,56 @@ public class TransactionContext {
 
         @Override
         public Statement createStatement(int resultSetType, int resultSetConcurrency, int resultSetHoldability) throws SQLException {
-            return delegate.createStatement(resultSetType, resultSetConcurrency, resultSetHoldability);
+            Statement statement = delegate.createStatement(resultSetType, resultSetConcurrency, resultSetHoldability);
+            if (queryTimeout > 0) {
+                statement.setQueryTimeout(queryTimeout);
+            }
+            return statement;
         }
 
         @Override
         public PreparedStatement prepareStatement(String sql, int resultSetType, int resultSetConcurrency, int resultSetHoldability) throws SQLException {
-            return delegate.prepareStatement(sql, resultSetType, resultSetConcurrency, resultSetHoldability);
+            PreparedStatement statement = delegate.prepareStatement(sql, resultSetType, resultSetConcurrency, resultSetHoldability);
+            if (queryTimeout > 0) {
+                statement.setQueryTimeout(queryTimeout);
+            }
+            return statement;
         }
 
         @Override
         public CallableStatement prepareCall(String sql, int resultSetType, int resultSetConcurrency, int resultSetHoldability) throws SQLException {
-            return delegate.prepareCall(sql, resultSetType, resultSetConcurrency, resultSetHoldability);
+            CallableStatement statement = delegate.prepareCall(sql, resultSetType, resultSetConcurrency, resultSetHoldability);
+            if (queryTimeout > 0) {
+                statement.setQueryTimeout(queryTimeout);
+            }
+            return statement;
         }
 
         @Override
         public PreparedStatement prepareStatement(String sql, int autoGeneratedKeys) throws SQLException {
-            return delegate.prepareStatement(sql, autoGeneratedKeys);
+            PreparedStatement statement = delegate.prepareStatement(sql, autoGeneratedKeys);
+            if (queryTimeout > 0) {
+                statement.setQueryTimeout(queryTimeout);
+            }
+            return statement;
         }
 
         @Override
         public PreparedStatement prepareStatement(String sql, int[] columnIndexes) throws SQLException {
-            return delegate.prepareStatement(sql, columnIndexes);
+            PreparedStatement statement = delegate.prepareStatement(sql, columnIndexes);
+            if (queryTimeout > 0) {
+                statement.setQueryTimeout(queryTimeout);
+            }
+            return statement;
         }
 
         @Override
         public PreparedStatement prepareStatement(String sql, String[] columnNames) throws SQLException {
-            return delegate.prepareStatement(sql, columnNames);
+            PreparedStatement statement = delegate.prepareStatement(sql, columnNames);
+            if (queryTimeout > 0) {
+                statement.setQueryTimeout(queryTimeout);
+            }
+            return statement;
         }
 
         @Override

activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4636Test.java

@@ -243,7 +243,7 @@ public class AMQ4636Test {
         public TestTransactionContext(
                 JDBCPersistenceAdapter jdbcPersistenceAdapter)
                 throws IOException {
-            super(jdbcPersistenceAdapter);
+            super(jdbcPersistenceAdapter, -1, -1);
         }
 
         @Override

activemq-unit-tests/src/test/java/org/apache/activemq/bugs/TrapMessageInJDBCStoreTest.java

@@ -277,7 +277,7 @@ public class TrapMessageInJDBCStoreTest extends TestCase {
         public TestTransactionContext(
                 JDBCPersistenceAdapter jdbcPersistenceAdapter)
                 throws IOException {
-            super(jdbcPersistenceAdapter);
+            super(jdbcPersistenceAdapter, -1, -1);
         }
 
         public void executeBatch() throws SQLException {

activemq-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JmsTransactionCommitFailureTest.java

@@ -335,7 +335,7 @@ public class JmsTransactionCommitFailureTest {
 
         @Override
         public TransactionContext getTransactionContext() throws IOException {
-            TransactionContext answer = new TransactionContext(this) {
+            TransactionContext answer = new TransactionContext(this, -1, -1) {
                 @Override
                 public void executeBatch() throws SQLException {
                     if (isCommitFailureEnabled) {

activemq-unit-tests/src/test/java/org/apache/activemq/store/jdbc/TransactionContextTest.java

@@ -41,7 +41,7 @@ public class TransactionContextTest {
 
     @Before
     public void setup() throws Exception {
-        underTest = new TransactionContext(jdbcPersistenceAdapter);
+        underTest = new TransactionContext(jdbcPersistenceAdapter, 0, 0);
     }
 
     @Test