Apache
/
activemq
mirror of https://github.com/apache/activemq.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

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

This commit is contained in:
gtully 2014-12-22 12:37:43 +00:00
parent d91bdc4cff
commit e16815ad3b
8 changed files with 112 additions and 37 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

66
activemq-broker/src/main/java/org/apache/activemq/util/LeaseLockerIOExceptionHandler.java Normal file
View File

@ -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;
}
}

4
activemq-jdbc-store/src/main/java/org/apache/activemq/store/jdbc/JDBCIOExceptionHandler.java
View File

@ -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);

4
activemq-unit-tests/src/test/java/org/apache/activemq/broker/ft/DbRestartJDBCQueueMasterSlaveLeaseQuiesceTest.java
View File

@ -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);
}

15
activemq-unit-tests/src/test/java/org/apache/activemq/broker/ft/DbRestartJDBCQueueMasterSlaveLeaseTest.java
View File

@ -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() {

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

@ -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;

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

@ -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;

3
activemq-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCIOExceptionHandlerMockeryTest.java
View File

@ -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 {

17
activemq-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCIOExceptionHandlerTest.java
View File

@ -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) {