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

Moving BrokerSubscriptionInfo processing into a new thread to prevent a
deadlock of the network bridge on startup
This commit is contained in:
Christopher L. Shannon (cshannon) 2016-10-24 07:51:34 -04:00
parent 1a811b72dd
commit b9cb02ae54
1 changed files with 51 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

74
activemq-broker/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java
View File

@ -163,6 +163,8 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br
protected BrokerService brokerService = null; protected BrokerService brokerService = null;
private ObjectName mbeanObjectName; private ObjectName mbeanObjectName;
private final ExecutorService serialExecutor = Executors.newSingleThreadExecutor(); private final ExecutorService serialExecutor = Executors.newSingleThreadExecutor();
//Use a new executor for processing BrokerSubscriptionInfo so we don't block other threads
private final ExecutorService syncExecutor = Executors.newSingleThreadExecutor();
private Transport duplexInboundLocalBroker = null; private Transport duplexInboundLocalBroker = null;
private ProducerInfo duplexInboundLocalProducerInfo; private ProducerInfo duplexInboundLocalProducerInfo;
@ -295,6 +297,14 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br
List<Runnable> pendingTasks = serialExecutor.shutdownNow(); List<Runnable> pendingTasks = serialExecutor.shutdownNow();
LOG.info("pending tasks on stop {}", pendingTasks); LOG.info("pending tasks on stop {}", pendingTasks);
} }
//Shutdown the syncExecutor, call countDown to make sure a thread can
//terminate if it is waiting
staticDestinationsLatch.countDown();
syncExecutor.shutdown();
if (!syncExecutor.awaitTermination(5, TimeUnit.SECONDS)) {
List<Runnable> pendingTasks = syncExecutor.shutdownNow();
LOG.info("pending tasks on stop {}", pendingTasks);
}
localBroker.oneway(new ShutdownInfo()); localBroker.oneway(new ShutdownInfo());
remoteBroker.oneway(new ShutdownInfo()); remoteBroker.oneway(new ShutdownInfo());
} catch (Throwable e) { } catch (Throwable e) {
@ -648,34 +658,52 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br
} else if (command.isBrokerInfo()) { } else if (command.isBrokerInfo()) {
futureRemoteBrokerInfo.set((BrokerInfo) command); futureRemoteBrokerInfo.set((BrokerInfo) command);
} else if (command instanceof BrokerSubscriptionInfo) { } else if (command instanceof BrokerSubscriptionInfo) {
staticDestinationsLatch.await(); final BrokerSubscriptionInfo brokerSubscriptionInfo = (BrokerSubscriptionInfo) command;
BrokerSubscriptionInfo subInfo = (BrokerSubscriptionInfo) command;
LOG.debug("Received Remote BrokerSubscriptionInfo on {} from {}",
this.brokerService.getBrokerName(), subInfo.getBrokerName());
if (configuration.isSyncDurableSubs() && configuration.isConduitSubscriptions() //Start in a new thread so we don't block the transport waiting for staticDestinations
&& !configuration.isDynamicOnly()) { syncExecutor.execute(new Runnable() {
if (started.get()) {
if (subInfo.getSubscriptionInfos() != null) { @Override
for (ConsumerInfo info : subInfo.getSubscriptionInfos()) { public void run() {
//re-add any process any non-NC consumers that match the try {
//dynamicallyIncludedDestinations list staticDestinationsLatch.await();
if((info.getSubscriptionName() == null || !info.getSubscriptionName().startsWith(DURABLE_SUB_PREFIX)) && //Make sure after the countDown of staticDestinationsLatch we aren't stopping
NetworkBridgeUtils.matchesDestinations(dynamicallyIncludedDestinations, info.getDestination())) { if (!disposed.get()) {
serviceRemoteConsumerAdvisory(info); BrokerSubscriptionInfo subInfo = brokerSubscriptionInfo;
LOG.debug("Received Remote BrokerSubscriptionInfo on {} from {}",
brokerService.getBrokerName(), subInfo.getBrokerName());
if (configuration.isSyncDurableSubs() && configuration.isConduitSubscriptions()
&& !configuration.isDynamicOnly()) {
if (started.get()) {
if (subInfo.getSubscriptionInfos() != null) {
for (ConsumerInfo info : subInfo.getSubscriptionInfos()) {
//re-add any process any non-NC consumers that match the
//dynamicallyIncludedDestinations list
if((info.getSubscriptionName() == null || !info.getSubscriptionName().startsWith(DURABLE_SUB_PREFIX)) &&
NetworkBridgeUtils.matchesDestinations(dynamicallyIncludedDestinations, info.getDestination())) {
serviceRemoteConsumerAdvisory(info);
}
}
}
//After re-added, clean up any empty durables
for (Iterator<DemandSubscription> i = subscriptionMapByLocalId.values().iterator(); i.hasNext(); ) {
DemandSubscription ds = i.next();
if (NetworkBridgeUtils.matchesDestinations(dynamicallyIncludedDestinations, ds.getLocalInfo().getDestination())) {
cleanupDurableSub(ds, i);
}
}
}
} }
} }
} } catch (Exception e) {
LOG.warn("Error processing BrokerSubscriptionInfo: {}", e.getMessage(), e);
//After re-added, clean up any empty durables LOG.debug(e.getMessage(), e);
for (Iterator<DemandSubscription> i = subscriptionMapByLocalId.values().iterator(); i.hasNext(); ) {
DemandSubscription ds = i.next();
if (NetworkBridgeUtils.matchesDestinations(dynamicallyIncludedDestinations, ds.getLocalInfo().getDestination())) {
cleanupDurableSub(ds, i);
}
} }
} }
} });
} else if (command.getClass() == ConnectionError.class) { } else if (command.getClass() == ConnectionError.class) {
ConnectionError ce = (ConnectionError) command; ConnectionError ce = (ConnectionError) command;
serviceRemoteException(ce.getException()); serviceRemoteException(ce.getException());