merge -c 835715 Fix for http://issues.apache.org/activemq/browse/AMQ-2439: IllegalState issue.

Tightening synchronization in DemandForwardingBridge to avoid subscription clean up and ack processing race condition.

git-svn-id: https://svn.apache.org/repos/asf/activemq/branches/activemq-5.3@835920 13f79535-47bb-0310-9956-ffa450edef68

activemq-core/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java

@@ -131,7 +131,6 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br
     private BrokerInfo localBrokerInfo;
     private BrokerInfo remoteBrokerInfo;
 
-
     private AtomicBoolean started = new AtomicBoolean();
     private TransportConnection duplexInitiatingConnection;
     private BrokerService brokerService = null;
@@ -675,7 +674,7 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br
                     enqueueCounter.incrementAndGet();
                     final MessageDispatch md = (MessageDispatch) command;
                     final DemandSubscription sub = subscriptionMapByLocalId.get(md.getConsumerId());
-                    if (sub != null && md.getMessage()!=null) {
+                    if (sub != null && md.getMessage() != null && sub.incrementOutstandingResponses()) {
                         // See if this consumer's brokerPath tells us it came from the broker at the other end
                         // of the bridge. I think we should be making this decision based on the message's
                         // broker bread crumbs and not the consumer's? However, the message's broker bread
@@ -697,6 +696,7 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br
                             // by bridging it using an async send (small chance
                             // of message loss).
 
+                            try {
                                 // Don't send it off to the remote if it originally came from the remote. 
                                 if (!cameFromRemote) {
                                     remoteBroker.oneway(message);
@@ -705,8 +705,12 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br
                                         LOG.debug("Message not forwarded on to remote, because message came from remote");
                                     }
                                 }
+
                                 localBroker.oneway(new MessageAck(md, MessageAck.INDIVIDUAL_ACK_TYPE, 1));
                                 dequeueCounter.incrementAndGet();
+                            } finally {
+                                sub.decrementOutstandingResponses();
+                            }
 
                         } else {
 
@@ -735,7 +739,6 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br
                             };
 
                             remoteBroker.asyncRequest(message, callback);
-                            sub.incrementOutstandingResponses();
                         }
 
                     } else {
@@ -1014,7 +1017,6 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br
         return suppress;
     }
 
- 
     private boolean hasLowerPriority(Subscription existingSub, ConsumerInfo candidateInfo) {
         boolean suppress = false;
 
@@ -1075,7 +1077,6 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br
         return doCreateDemandSubscription(info);
     }
 
-    
     protected DemandSubscription doCreateDemandSubscription(ConsumerInfo info) throws IOException {
         DemandSubscription result = new DemandSubscription(info);
         result.getLocalInfo().setConsumerId(new ConsumerId(localSessionInfo.getSessionId(), consumerIdGenerator.getNextSequenceId()));
@@ -1132,7 +1133,6 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br
         sub.getLocalInfo().setAdditionalPredicate(createNetworkBridgeFilter(info));
     }
 
-    
     protected void removeDemandSubscription(ConsumerId id) throws IOException {
         DemandSubscription sub = subscriptionMapByRemoteId.remove(id);
         if (LOG.isDebugEnabled()) {
@@ -1178,7 +1178,6 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br
 
     protected abstract void serviceRemoteBrokerInfo(Command command) throws IOException;
 
-
     protected abstract BrokerId[] getRemoteBrokerPath();
 
     public void setNetworkBridgeListener(NetworkBridgeListener listener) {

activemq-core/src/main/java/org/apache/activemq/network/DemandSubscription.java

@@ -37,6 +37,7 @@ public class DemandSubscription {
     private final ConsumerInfo remoteInfo;
     private final ConsumerInfo localInfo;
     private Set<ConsumerId> remoteSubsIds = new CopyOnWriteArraySet<ConsumerId>();
+
     private AtomicInteger dispatched = new AtomicInteger(0);
     private AtomicBoolean activeWaiter = new AtomicBoolean();
 
@@ -81,7 +82,6 @@ public class DemandSubscription {
         return localInfo;
     }
 
-    
     /**
      * @return Returns the remoteInfo.
      */
@@ -117,7 +117,12 @@ public class DemandSubscription {
         }
     }
 
-    public void incrementOutstandingResponses() {
+    public boolean incrementOutstandingResponses() {
         dispatched.incrementAndGet();
+        if (activeWaiter.get()) {
+            decrementOutstandingResponses();
+            return false;
+        }
+        return true;
     }
 }

activemq-core/src/main/java/org/apache/activemq/transport/ResponseCorrelator.java

@@ -65,12 +65,19 @@ public class ResponseCorrelator extends TransportFilter {
         command.setCommandId(sequenceGenerator.getNextSequenceId());
         command.setResponseRequired(true);
         FutureResponse future = new FutureResponse(responseCallback);
+        IOException priorError = null;
         synchronized (requestMap) {
-            if( this.error !=null ) {
+            priorError = this.error;
-                throw error;
+            if (priorError == null) {
-            }
                 requestMap.put(new Integer(command.getCommandId()), future);
             }
+        }
+
+        if (priorError != null) {
+            future.set(new ExceptionResponse(priorError));
+            throw priorError;
+        }
+
         next.oneway(command);
         return future;
     }