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-6248 

Prevent conccurent calls to handleTransportFailure from closing an
already reconnected transport instance.
This commit is contained in:
Timothy Bish 2016-04-18 09:45:45 -04:00
parent 11622b3af3
commit 23a5beb86c
1 changed files with 86 additions and 109 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

141
activemq-client/src/main/java/org/apache/activemq/transport/failover/FailoverTransport.java
View File

@ -111,9 +111,7 @@ public class FailoverTransport implements CompositeTransport {
private boolean trackMessages = false; private boolean trackMessages = false;
private boolean trackTransactionProducers = true; private boolean trackTransactionProducers = true;
private int maxCacheSize = 128 * 1024; private int maxCacheSize = 128 * 1024;
private final TransportListener disposedListener = new DefaultTransportListener() { private final TransportListener disposedListener = new DefaultTransportListener() {};
};
private final TransportListener myTransportListener = createTransportListener();
private boolean updateURIsSupported = true; private boolean updateURIsSupported = true;
private boolean reconnectSupported = true; private boolean reconnectSupported = true;
// remember for reconnect thread // remember for reconnect thread
@ -180,11 +178,8 @@ public class FailoverTransport implements CompositeTransport {
}, "ActiveMQ Failover Worker: " + System.identityHashCode(this)); }, "ActiveMQ Failover Worker: " + System.identityHashCode(this));
} }
TransportListener createTransportListener() { private void processCommand(Object incoming) {
return new TransportListener() { Command command = (Command) incoming;
@Override
public void onCommand(Object o) {
Command command = (Command) o;
if (command == null) { if (command == null) {
return; return;
} }
@ -200,41 +195,44 @@ public class FailoverTransport implements CompositeTransport {
if (command.isConnectionControl()) { if (command.isConnectionControl()) {
handleConnectionControl((ConnectionControl) command); handleConnectionControl((ConnectionControl) command);
} } else if (command.isConsumerControl()) {
else if (command.isConsumerControl()) {
ConsumerControl consumerControl = (ConsumerControl)command; ConsumerControl consumerControl = (ConsumerControl)command;
if (consumerControl.isClose()) { if (consumerControl.isClose()) {
stateTracker.processRemoveConsumer(consumerControl.getConsumerId(), RemoveInfo.LAST_DELIVERED_UNKNOWN); stateTracker.processRemoveConsumer(consumerControl.getConsumerId(), RemoveInfo.LAST_DELIVERED_UNKNOWN);
} }
} }
if (transportListener != null) { if (transportListener != null) {
transportListener.onCommand(command); transportListener.onCommand(command);
} }
} }
private TransportListener createTransportListener(final Transport owner) {
return new TransportListener() {
@Override
public void onCommand(Object o) {
processCommand(o);
}
@Override @Override
public void onException(IOException error) { public void onException(IOException error) {
try { try {
handleTransportFailure(error); handleTransportFailure(owner, error);
} catch (InterruptedException e) { } catch (InterruptedException e) {
Thread.currentThread().interrupt(); Thread.currentThread().interrupt();
if (transportListener != null) {
transportListener.onException(new InterruptedIOException()); transportListener.onException(new InterruptedIOException());
} }
} }
}
@Override @Override
public void transportInterupted() { public void transportInterupted() {
if (transportListener != null) {
transportListener.transportInterupted();
}
} }
@Override @Override
public void transportResumed() { public void transportResumed() {
if (transportListener != null) {
transportListener.transportResumed();
}
} }
}; };
} }
@ -245,6 +243,10 @@ public class FailoverTransport implements CompositeTransport {
} }
public final void handleTransportFailure(IOException e) throws InterruptedException { public final void handleTransportFailure(IOException e) throws InterruptedException {
handleTransportFailure(getConnectedTransport(), e);
}
public final void handleTransportFailure(Transport failed, IOException e) throws InterruptedException {
if (shuttingDown) { if (shuttingDown) {
// shutdown info sent and remote socket closed and we see that before a local close // shutdown info sent and remote socket closed and we see that before a local close
// let the close do the work // let the close do the work
@ -256,21 +258,25 @@ public class FailoverTransport implements CompositeTransport {
} }
// could be blocked in write with the reconnectMutex held, but still needs to be whacked // could be blocked in write with the reconnectMutex held, but still needs to be whacked
Transport transport = connectedTransport.getAndSet(null); Transport transport = null;
if (connectedTransport.compareAndSet(failed, null)) {
transport = failed;
if (transport != null) { if (transport != null) {
disposeTransport(transport); disposeTransport(transport);
} }
}
synchronized (reconnectMutex) { synchronized (reconnectMutex) {
if (transport != null && connectedTransport.get() == null) { if (transport != null && connectedTransport.get() == null) {
boolean reconnectOk = false; boolean reconnectOk = false;
if (canReconnect()) { if (canReconnect()) {
reconnectOk = true; reconnectOk = true;
} }
LOG.warn("Transport (" + connectedTransportURI + ") failed"
+ (reconnectOk ? "," : ", not") + " attempting to automatically reconnect", e); LOG.warn("Transport ({}) failed {} attempting to automatically reconnect: {}",
connectedTransport, (reconnectOk ? "," : ", not"), e);
failedConnectTransportURI = connectedTransportURI; failedConnectTransportURI = connectedTransportURI;
connectedTransportURI = null; connectedTransportURI = null;
@ -347,9 +353,7 @@ public class FailoverTransport implements CompositeTransport {
@Override @Override
public void start() throws Exception { public void start() throws Exception {
synchronized (reconnectMutex) { synchronized (reconnectMutex) {
if (LOG.isDebugEnabled()) { LOG.debug("Started {}", this);
LOG.debug("Started " + this);
}
if (started) { if (started) {
return; return;
} }
@ -373,7 +377,7 @@ public class FailoverTransport implements CompositeTransport {
try { try {
synchronized (reconnectMutex) { synchronized (reconnectMutex) {
if (LOG.isDebugEnabled()) { if (LOG.isDebugEnabled()) {
LOG.debug("Stopped " + this); LOG.debug("Stopped {}", this);
} }
if (!started) { if (!started) {
return; return;
@ -407,9 +411,7 @@ public class FailoverTransport implements CompositeTransport {
} }
for (Transport transport : backupsToStop) { for (Transport transport : backupsToStop) {
try { try {
if (LOG.isTraceEnabled()) { LOG.trace("Stopped backup: {}", transport);
LOG.trace("Stopped backup: " + transport);
}
disposeTransport(transport); disposeTransport(transport);
} catch (Exception e) { } catch (Exception e) {
} }
@ -581,7 +583,7 @@ public class FailoverTransport implements CompositeTransport {
if (command.isResponseRequired()) { if (command.isResponseRequired()) {
Response response = new Response(); Response response = new Response();
response.setCorrelationId(command.getCommandId()); response.setCorrelationId(command.getCommandId());
myTransportListener.onCommand(response); processCommand(response);
} }
return; return;
} else if (command instanceof MessagePull) { } else if (command instanceof MessagePull) {
@ -591,7 +593,7 @@ public class FailoverTransport implements CompositeTransport {
MessageDispatch dispatch = new MessageDispatch(); MessageDispatch dispatch = new MessageDispatch();
dispatch.setConsumerId(pullRequest.getConsumerId()); dispatch.setConsumerId(pullRequest.getConsumerId());
dispatch.setDestination(pullRequest.getDestination()); dispatch.setDestination(pullRequest.getDestination());
myTransportListener.onCommand(dispatch); processCommand(dispatch);
} }
return; return;
} }
@ -607,24 +609,19 @@ public class FailoverTransport implements CompositeTransport {
boolean timedout = false; boolean timedout = false;
while (transport == null && !disposed && connectionFailure == null while (transport == null && !disposed && connectionFailure == null
&& !Thread.currentThread().isInterrupted() && willReconnect()) { && !Thread.currentThread().isInterrupted() && willReconnect()) {
if (LOG.isTraceEnabled()) {
LOG.trace("Waiting for transport to reconnect..: " + command); LOG.trace("Waiting for transport to reconnect..: {}", command);
}
long end = System.currentTimeMillis(); long end = System.currentTimeMillis();
if (command.isMessage() && timeout > 0 && (end - start > timeout)) { if (command.isMessage() && timeout > 0 && (end - start > timeout)) {
timedout = true; timedout = true;
if (LOG.isInfoEnabled()) { LOG.info("Failover timed out after {} ms", (end - start));
LOG.info("Failover timed out after " + (end - start) + "ms");
}
break; break;
} }
try { try {
reconnectMutex.wait(100); reconnectMutex.wait(100);
} catch (InterruptedException e) { } catch (InterruptedException e) {
Thread.currentThread().interrupt(); Thread.currentThread().interrupt();
if (LOG.isDebugEnabled()) { LOG.debug("Interupted:", e);
LOG.debug("Interupted: " + e, e);
}
} }
transport = connectedTransport.get(); transport = connectedTransport.get();
} }
@ -650,7 +647,7 @@ public class FailoverTransport implements CompositeTransport {
try { try {
tracked = stateTracker.track(command); tracked = stateTracker.track(command);
} catch (IOException ioe) { } catch (IOException ioe) {
LOG.debug("Cannot track the command " + command, ioe); LOG.debug("Cannot track the command {} {}", command, ioe);
} }
// If it was a request and it was not being tracked by // If it was a request and it was not being tracked by
// the state tracker, // the state tracker,
@ -691,19 +688,14 @@ public class FailoverTransport implements CompositeTransport {
} else { } else {
// Handle the error but allow the method to return since the // Handle the error but allow the method to return since the
// tracked commands are replayed on reconnect. // tracked commands are replayed on reconnect.
if (LOG.isDebugEnabled()) { LOG.debug("Send oneway attempt: {} failed for command: {}", i, command);
LOG.debug("Send oneway attempt: " + i + " failed for command:" + command);
}
handleTransportFailure(e); handleTransportFailure(e);
} }
} }
return; return;
} catch (IOException e) { } catch (IOException e) {
if (LOG.isDebugEnabled()) { LOG.debug("Send oneway attempt: {} failed for command: {}", i, command);
LOG.debug("Send oneway attempt: " + i + " failed for command:" + command);
}
handleTransportFailure(e); handleTransportFailure(e);
} }
} }
@ -774,7 +766,7 @@ public class FailoverTransport implements CompositeTransport {
} }
} catch (Exception e) { } catch (Exception e) {
LOG.error("Failed to parse URI: " + u); LOG.error("Failed to parse URI: {}", u);
} }
} }
@ -818,9 +810,9 @@ public class FailoverTransport implements CompositeTransport {
if (removed) { if (removed) {
l.add(failedConnectTransportURI); l.add(failedConnectTransportURI);
} }
if (LOG.isDebugEnabled()) {
LOG.debug("urlList connectionList:" + l + ", from: " + uris); LOG.debug("urlList connectionList:{}, from: {}", l, uris);
}
return l; return l;
} }
@ -861,9 +853,7 @@ public class FailoverTransport implements CompositeTransport {
tmpMap = new LinkedHashMap<Integer, Command>(requestMap); tmpMap = new LinkedHashMap<Integer, Command>(requestMap);
} }
for (Command command : tmpMap.values()) { for (Command command : tmpMap.values()) {
if (LOG.isTraceEnabled()) { LOG.trace("restore requestMap, replay: {}", command);
LOG.trace("restore requestMap, replay: " + command);
}
t.oneway(command); t.oneway(command);
} }
} }
@ -916,7 +906,7 @@ public class FailoverTransport implements CompositeTransport {
} }
newUris = buffer.toString(); newUris = buffer.toString();
} catch (IOException ioe) { } catch (IOException ioe) {
LOG.error("Failed to read updateURIsURL: " + fileURL, ioe); LOG.error("Failed to read updateURIsURL: {} {}",fileURL, ioe);
} finally { } finally {
if (in != null) { if (in != null) {
try { try {
@ -954,9 +944,7 @@ public class FailoverTransport implements CompositeTransport {
doRebalance = false; doRebalance = false;
return false; return false;
} else { } else {
if (LOG.isDebugEnabled()) { LOG.debug("Doing rebalance from: {} to {}", connectedTransportURI, connectList);
LOG.debug("Doing rebalance from: " + connectedTransportURI + " to " + connectList);
}
try { try {
Transport transport = this.connectedTransport.getAndSet(null); Transport transport = this.connectedTransport.getAndSet(null);
@ -964,11 +952,9 @@ public class FailoverTransport implements CompositeTransport {
disposeTransport(transport); disposeTransport(transport);
} }
} catch (Exception e) { } catch (Exception e) {
if (LOG.isDebugEnabled()) {
LOG.debug("Caught an exception stopping existing transport for rebalance", e); LOG.debug("Caught an exception stopping existing transport for rebalance", e);
} }
} }
}
doRebalance = false; doRebalance = false;
} }
@ -988,7 +974,7 @@ public class FailoverTransport implements CompositeTransport {
backups.remove(bt); backups.remove(bt);
transport = bt.getTransport(); transport = bt.getTransport();
uri = bt.getUri(); uri = bt.getUri();
myTransportListener.onCommand(bt.getBrokerInfo()); processCommand(bt.getBrokerInfo());
if (priorityBackup && priorityBackupAvailable) { if (priorityBackup && priorityBackupAvailable) {
Transport old = this.connectedTransport.getAndSet(null); Transport old = this.connectedTransport.getAndSet(null);
if (old != null) { if (old != null) {
@ -1023,19 +1009,17 @@ public class FailoverTransport implements CompositeTransport {
transport = TransportFactory.compositeConnect(uri); transport = TransportFactory.compositeConnect(uri);
} }
if (LOG.isDebugEnabled()) { LOG.debug("Attempting {}th connect to: {}", connectFailures, uri);
LOG.debug("Attempting " + connectFailures + "th connect to: " + uri);
} transport.setTransportListener(createTransportListener(transport));
transport.setTransportListener(myTransportListener);
transport.start(); transport.start();
if (started && !firstConnection) { if (started && !firstConnection) {
restoreTransport(transport); restoreTransport(transport);
} }
if (LOG.isDebugEnabled()) {
LOG.debug("Connection established"); LOG.debug("Connection established");
}
reconnectDelay = initialReconnectDelay; reconnectDelay = initialReconnectDelay;
connectedTransportURI = uri; connectedTransportURI = uri;
connectedTransport.set(transport); connectedTransport.set(transport);
@ -1058,33 +1042,26 @@ public class FailoverTransport implements CompositeTransport {
if (transportListener != null) { if (transportListener != null) {
transportListener.transportResumed(); transportListener.transportResumed();
} else { } else {
if (LOG.isDebugEnabled()) {
LOG.debug("transport resumed by transport listener not set"); LOG.debug("transport resumed by transport listener not set");
} }
}
if (firstConnection) { if (firstConnection) {
firstConnection = false; firstConnection = false;
LOG.info("Successfully connected to " + uri); LOG.info("Successfully connected to {}", uri);
} else { } else {
LOG.info("Successfully reconnected to " + uri); LOG.info("Successfully reconnected to {}", uri);
} }
return false; return false;
} catch (Exception e) { } catch (Exception e) {
failure = e; failure = e;
if (LOG.isDebugEnabled()) { LOG.debug("Connect fail to: {}, reason: {}", uri, e);
LOG.debug("Connect fail to: " + uri + ", reason: " + e);
}
if (transport != null) { if (transport != null) {
try { try {
transport.stop(); transport.stop();
transport = null; transport = null;
} catch (Exception ee) { } catch (Exception ee) {
if (LOG.isDebugEnabled()) { LOG.debug("Stop of failed transport: {} failed with reason: {}", transport, ee);
LOG.debug("Stop of failed transport: " + transport +
" failed with reason: " + ee);
}
} }
} }
} finally { } finally {
@ -1098,7 +1075,7 @@ public class FailoverTransport implements CompositeTransport {
connectFailures++; connectFailures++;
if (reconnectLimit != INFINITE && connectFailures >= reconnectLimit) { if (reconnectLimit != INFINITE && connectFailures >= reconnectLimit) {
LOG.error("Failed to connect to " + uris + " after: " + connectFailures + " attempt(s)"); LOG.error("Failed to connect to {} after: {} attempt(s)", uris, connectFailures);
connectionFailure = failure; connectionFailure = failure;
// Make sure on initial startup, that the transportListener has been // Make sure on initial startup, that the transportListener has been
@ -1406,9 +1383,9 @@ public class FailoverTransport implements CompositeTransport {
} catch(IOException e) { } catch(IOException e) {
if (firstAddr == null) { if (firstAddr == null) {
LOG.error("Failed to Lookup INetAddress for URI[ " + first + " ] : " + e); LOG.error("Failed to Lookup INetAddress for URI[{}] : {}", first, e);
} else { } else {
LOG.error("Failed to Lookup INetAddress for URI[ " + second + " ] : " + e); LOG.error("Failed to Lookup INetAddress for URI[{}] : {}", second, e);
} }
if (first.getHost().equalsIgnoreCase(second.getHost())) { if (first.getHost().equalsIgnoreCase(second.getHost())) {