Repository: activemq Updated Branches: refs/heads/trunk d91bdc4cf -> e16815ad3
https://issues.apache.org/jira/browse/AMQ-5174 - apply patch from paulGale with thanks replace jdbc specific lease io handler with generic lease io handler - closesThis closes #53 Project: http://git-wip-us.apache.org/repos/asf/activemq/repo Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/e16815ad Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/e16815ad Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/e16815ad Branch: refs/heads/trunk Commit: e16815ad3bae86f4e5c970db3c4647793d185416 Parents: d91bdc4 Author: gtully <[email protected]> Authored: Mon Dec 22 12:37:43 2014 +0000 Committer: gtully <[email protected]> Committed: Mon Dec 22 12:37:43 2014 +0000 ---------------------------------------------------------------------- .../util/LeaseLockerIOExceptionHandler.java | 66 ++++++++++++++++++++ .../store/jdbc/JDBCIOExceptionHandler.java | 4 ++ ...artJDBCQueueMasterSlaveLeaseQuiesceTest.java | 4 +- .../DbRestartJDBCQueueMasterSlaveLeaseTest.java | 15 ++--- .../org/apache/activemq/bugs/AMQ4636Test.java | 4 +- .../bugs/TrapMessageInJDBCStoreTest.java | 36 ++++++----- .../jdbc/JDBCIOExceptionHandlerMockeryTest.java | 3 +- .../store/jdbc/JDBCIOExceptionHandlerTest.java | 17 ++--- 8 files changed, 112 insertions(+), 37 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/activemq/blob/e16815ad/activemq-broker/src/main/java/org/apache/activemq/util/LeaseLockerIOExceptionHandler.java ---------------------------------------------------------------------- diff --git a/activemq-broker/src/main/java/org/apache/activemq/util/LeaseLockerIOExceptionHandler.java b/activemq-broker/src/main/java/org/apache/activemq/util/LeaseLockerIOExceptionHandler.java new file mode 100644 index 0000000..bd17e79 --- /dev/null +++ b/activemq-broker/src/main/java/org/apache/activemq/util/LeaseLockerIOExceptionHandler.java @@ -0,0 +1,66 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.activemq.util; + +import org.apache.activemq.broker.LockableServiceSupport; +import org.apache.activemq.broker.Locker; +import org.apache.activemq.broker.SuppressReplyException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; + +/** + * @org.apache.xbean.XBean + */ +public class LeaseLockerIOExceptionHandler extends DefaultIOExceptionHandler { + private static final Logger LOG = LoggerFactory.getLogger(LeaseLockerIOExceptionHandler.class); + + public LeaseLockerIOExceptionHandler() { + setIgnoreSQLExceptions(false); + setStopStartConnectors(true); + } + + // fail only when we get an authoritative answer from the db w/o exceptions + @Override + protected boolean hasLockOwnership() throws IOException { + boolean hasLock = true; + + if (broker.getPersistenceAdapter() instanceof LockableServiceSupport) { + Locker locker = ((LockableServiceSupport) broker.getPersistenceAdapter()).getLocker(); + + if (locker != null) { + try { + if (!locker.keepAlive()) { + hasLock = false; + } + } + catch (SuppressReplyException ignoreWhileHandlingInProgress) { + } + catch (IOException ignored) { + } + + if (!hasLock) { + LOG.warn("Lock keepAlive failed, no longer lock owner with: {}", locker); + throw new IOException("Lock keepAlive failed, no longer lock owner with: " + locker); + } + } + } + + return hasLock; + } +} http://git-wip-us.apache.org/repos/asf/activemq/blob/e16815ad/activemq-jdbc-store/src/main/java/org/apache/activemq/store/jdbc/JDBCIOExceptionHandler.java ---------------------------------------------------------------------- diff --git a/activemq-jdbc-store/src/main/java/org/apache/activemq/store/jdbc/JDBCIOExceptionHandler.java b/activemq-jdbc-store/src/main/java/org/apache/activemq/store/jdbc/JDBCIOExceptionHandler.java index d0ea276..8117887 100644 --- a/activemq-jdbc-store/src/main/java/org/apache/activemq/store/jdbc/JDBCIOExceptionHandler.java +++ b/activemq-jdbc-store/src/main/java/org/apache/activemq/store/jdbc/JDBCIOExceptionHandler.java @@ -27,6 +27,10 @@ import org.slf4j.LoggerFactory; /** * @org.apache.xbean.XBean */ +/* + * @deprecated Use more general {@link org.apache.activemq.util.LeaseLockerIOExceptionHandler} instead + */ +@Deprecated public class JDBCIOExceptionHandler extends DefaultIOExceptionHandler { private static final Logger LOG = LoggerFactory.getLogger(JDBCIOExceptionHandler.class); http://git-wip-us.apache.org/repos/asf/activemq/blob/e16815ad/activemq-unit-tests/src/test/java/org/apache/activemq/broker/ft/DbRestartJDBCQueueMasterSlaveLeaseQuiesceTest.java ---------------------------------------------------------------------- diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/broker/ft/DbRestartJDBCQueueMasterSlaveLeaseQuiesceTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/broker/ft/DbRestartJDBCQueueMasterSlaveLeaseQuiesceTest.java index 228f9c0..e71cfe6 100644 --- a/activemq-unit-tests/src/test/java/org/apache/activemq/broker/ft/DbRestartJDBCQueueMasterSlaveLeaseQuiesceTest.java +++ b/activemq-unit-tests/src/test/java/org/apache/activemq/broker/ft/DbRestartJDBCQueueMasterSlaveLeaseQuiesceTest.java @@ -19,7 +19,7 @@ package org.apache.activemq.broker.ft; import java.util.concurrent.TimeUnit; import org.apache.activemq.ActiveMQConnection; import org.apache.activemq.broker.BrokerService; -import org.apache.activemq.store.jdbc.JDBCIOExceptionHandler; +import org.apache.activemq.util.LeaseLockerIOExceptionHandler; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -31,7 +31,7 @@ public class DbRestartJDBCQueueMasterSlaveLeaseQuiesceTest extends DbRestartJDBC @Override protected void configureBroker(BrokerService brokerService) { // master and slave survive db restart and retain master/slave status - JDBCIOExceptionHandler stopConnectors = new JDBCIOExceptionHandler(); + LeaseLockerIOExceptionHandler stopConnectors = new LeaseLockerIOExceptionHandler(); brokerService.setIoExceptionHandler(stopConnectors); } http://git-wip-us.apache.org/repos/asf/activemq/blob/e16815ad/activemq-unit-tests/src/test/java/org/apache/activemq/broker/ft/DbRestartJDBCQueueMasterSlaveLeaseTest.java ---------------------------------------------------------------------- diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/broker/ft/DbRestartJDBCQueueMasterSlaveLeaseTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/broker/ft/DbRestartJDBCQueueMasterSlaveLeaseTest.java index 9f51ab7..cf4929a 100644 --- a/activemq-unit-tests/src/test/java/org/apache/activemq/broker/ft/DbRestartJDBCQueueMasterSlaveLeaseTest.java +++ b/activemq-unit-tests/src/test/java/org/apache/activemq/broker/ft/DbRestartJDBCQueueMasterSlaveLeaseTest.java @@ -17,14 +17,11 @@ package org.apache.activemq.broker.ft; import java.io.IOException; -import java.util.concurrent.TimeUnit; -import org.apache.activemq.ActiveMQConnection; import org.apache.activemq.broker.BrokerService; -import org.apache.activemq.store.jdbc.JDBCIOExceptionHandler; import org.apache.activemq.store.jdbc.JDBCPersistenceAdapter; import org.apache.activemq.store.jdbc.LeaseDatabaseLocker; -import org.apache.activemq.util.DefaultIOExceptionHandler; +import org.apache.activemq.util.LeaseLockerIOExceptionHandler; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -43,11 +40,11 @@ public class DbRestartJDBCQueueMasterSlaveLeaseTest extends DbRestartJDBCQueueMa protected void configureBroker(BrokerService brokerService) { //let the brokers die on exception and master should have lease on restart // which will delay slave start till it expires - JDBCIOExceptionHandler trapSQLExceptions = new JDBCIOExceptionHandler(); - trapSQLExceptions.setIgnoreSQLExceptions(false); - trapSQLExceptions.setStopStartConnectors(false); - trapSQLExceptions.setResumeCheckSleepPeriod(500l); - brokerService.setIoExceptionHandler(trapSQLExceptions); + LeaseLockerIOExceptionHandler ioExceptionHandler = new LeaseLockerIOExceptionHandler(); + ioExceptionHandler.setIgnoreSQLExceptions(false); + ioExceptionHandler.setStopStartConnectors(false); + ioExceptionHandler.setResumeCheckSleepPeriod(500l); + brokerService.setIoExceptionHandler(ioExceptionHandler); } private long getLockKeepAlivePeriod() { http://git-wip-us.apache.org/repos/asf/activemq/blob/e16815ad/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4636Test.java ---------------------------------------------------------------------- diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4636Test.java b/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4636Test.java index 4373d49..014d86a 100644 --- a/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4636Test.java +++ b/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4636Test.java @@ -33,11 +33,11 @@ import org.apache.activemq.broker.BrokerService; import org.apache.activemq.broker.region.policy.PolicyEntry; import org.apache.activemq.broker.region.policy.PolicyMap; import org.apache.activemq.store.jdbc.DataSourceServiceSupport; -import org.apache.activemq.store.jdbc.JDBCIOExceptionHandler; import org.apache.activemq.store.jdbc.JDBCPersistenceAdapter; import org.apache.activemq.store.jdbc.LeaseDatabaseLocker; import org.apache.activemq.store.jdbc.TransactionContext; import org.apache.activemq.util.IOHelper; +import org.apache.activemq.util.LeaseLockerIOExceptionHandler; import org.apache.derby.jdbc.EmbeddedDataSource; import org.junit.After; import org.junit.Before; @@ -114,7 +114,7 @@ public class AMQ4636Test { broker.setDestinationPolicy(policyMap); broker.setPersistenceAdapter(jdbc); - broker.setIoExceptionHandler(new JDBCIOExceptionHandler()); + broker.setIoExceptionHandler(new LeaseLockerIOExceptionHandler()); transportUrl = broker.addConnector(transportUrl).getPublishableConnectString(); return broker; http://git-wip-us.apache.org/repos/asf/activemq/blob/e16815ad/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/TrapMessageInJDBCStoreTest.java ---------------------------------------------------------------------- diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/TrapMessageInJDBCStoreTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/TrapMessageInJDBCStoreTest.java index 2fedbeb..0d14135 100644 --- a/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/TrapMessageInJDBCStoreTest.java +++ b/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/TrapMessageInJDBCStoreTest.java @@ -16,29 +16,35 @@ */ package org.apache.activemq.bugs; -import junit.framework.TestCase; -import org.apache.activemq.ActiveMQConnectionFactory; -import org.apache.activemq.broker.BrokerService; -import org.apache.activemq.command.*; -import org.apache.activemq.store.jdbc.*; -import org.apache.activemq.util.ByteSequence; -import org.apache.activemq.util.IOHelper; -import org.apache.derby.jdbc.EmbeddedDataSource; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import javax.jms.*; -import javax.jms.Message; import java.io.IOException; import java.sql.PreparedStatement; import java.sql.ResultSet; import java.sql.SQLException; import java.util.ArrayList; import java.util.List; -import java.util.concurrent.Executor; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; +import javax.jms.Connection; +import javax.jms.DeliveryMode; +import javax.jms.Destination; +import javax.jms.JMSException; +import javax.jms.MessageConsumer; +import javax.jms.MessageProducer; +import javax.jms.Session; +import javax.jms.TextMessage; +import junit.framework.TestCase; +import org.apache.activemq.ActiveMQConnectionFactory; +import org.apache.activemq.broker.BrokerService; +import org.apache.activemq.store.jdbc.DataSourceServiceSupport; +import org.apache.activemq.store.jdbc.JDBCPersistenceAdapter; +import org.apache.activemq.store.jdbc.LeaseDatabaseLocker; +import org.apache.activemq.store.jdbc.TransactionContext; +import org.apache.activemq.util.IOHelper; +import org.apache.activemq.util.LeaseLockerIOExceptionHandler; +import org.apache.derby.jdbc.EmbeddedDataSource; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** * Test to demostrate a message trapped in the JDBC store and not @@ -82,7 +88,7 @@ public class TrapMessageInJDBCStoreTest extends TestCase { broker.setPersistenceAdapter(jdbc); - broker.setIoExceptionHandler(new JDBCIOExceptionHandler()); + broker.setIoExceptionHandler(new LeaseLockerIOExceptionHandler()); transportUrl = broker.addConnector(transportUrl).getPublishableConnectString(); return broker; http://git-wip-us.apache.org/repos/asf/activemq/blob/e16815ad/activemq-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCIOExceptionHandlerMockeryTest.java ---------------------------------------------------------------------- diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCIOExceptionHandlerMockeryTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCIOExceptionHandlerMockeryTest.java index dd5d506..39ebf50 100644 --- a/activemq-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCIOExceptionHandlerMockeryTest.java +++ b/activemq-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCIOExceptionHandlerMockeryTest.java @@ -21,6 +21,7 @@ import java.util.HashMap; import org.apache.activemq.broker.BrokerService; import org.apache.activemq.broker.Locker; import org.apache.activemq.broker.SuppressReplyException; +import org.apache.activemq.util.LeaseLockerIOExceptionHandler; import org.apache.activemq.util.ServiceStopper; import org.apache.activemq.util.Wait; import org.jmock.Expectations; @@ -85,7 +86,7 @@ public class JDBCIOExceptionHandlerMockeryTest { }}); - JDBCIOExceptionHandler underTest = new JDBCIOExceptionHandler(); + LeaseLockerIOExceptionHandler underTest = new LeaseLockerIOExceptionHandler(); underTest.setBrokerService(brokerService); try { http://git-wip-us.apache.org/repos/asf/activemq/blob/e16815ad/activemq-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCIOExceptionHandlerTest.java ---------------------------------------------------------------------- diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCIOExceptionHandlerTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCIOExceptionHandlerTest.java index c56bc7f..d0ac1da 100644 --- a/activemq-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCIOExceptionHandlerTest.java +++ b/activemq-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCIOExceptionHandlerTest.java @@ -28,6 +28,7 @@ import junit.framework.TestCase; import org.apache.activemq.ActiveMQConnectionFactory; import org.apache.activemq.broker.BrokerService; +import org.apache.activemq.util.LeaseLockerIOExceptionHandler; import org.apache.activemq.util.Wait; import org.apache.derby.jdbc.EmbeddedDataSource; import org.slf4j.Logger; @@ -78,10 +79,10 @@ public class JDBCIOExceptionHandlerTest extends TestCase { } broker.setPersistenceAdapter(jdbc); - JDBCIOExceptionHandler jdbcioExceptionHandler = new JDBCIOExceptionHandler(); - jdbcioExceptionHandler.setResumeCheckSleepPeriod(1000l); - jdbcioExceptionHandler.setStopStartConnectors(startStopConnectors); - broker.setIoExceptionHandler(jdbcioExceptionHandler); + LeaseLockerIOExceptionHandler ioExceptionHandler = new LeaseLockerIOExceptionHandler(); + ioExceptionHandler.setResumeCheckSleepPeriod(1000l); + ioExceptionHandler.setStopStartConnectors(startStopConnectors); + broker.setIoExceptionHandler(ioExceptionHandler); String connectionUri = broker.addConnector(TRANSPORT_URL).getPublishableConnectString(); factory = new ActiveMQConnectionFactory(connectionUri); @@ -137,10 +138,10 @@ public class JDBCIOExceptionHandlerTest extends TestCase { } broker.setPersistenceAdapter(jdbc); - JDBCIOExceptionHandler jdbcioExceptionHandler = new JDBCIOExceptionHandler(); - jdbcioExceptionHandler.setResumeCheckSleepPeriod(1000l); - jdbcioExceptionHandler.setStopStartConnectors(false); - broker.setIoExceptionHandler(jdbcioExceptionHandler); + LeaseLockerIOExceptionHandler ioExceptionHandler = new LeaseLockerIOExceptionHandler(); + ioExceptionHandler.setResumeCheckSleepPeriod(1000l); + ioExceptionHandler.setStopStartConnectors(false); + broker.setIoExceptionHandler(ioExceptionHandler); slave.set(broker); broker.start(); } catch (Exception e) {
