- Gaurd access to dispatched list ( a sync was missing).

- Added better exception messages to know what happened when a slave subscription gets out of sync with the master.
- Implemented a simpler isFull()

git-svn-id: https://svn.apache.org/repos/asf/incubator/activemq/trunk@398015 13f79535-47bb-0310-9956-ffa450edef68

activemq-core/src/main/java/org/apache/activemq/broker/region/PrefetchSubscription.java

@@ -60,13 +60,12 @@ abstract public class PrefetchSubscription extends AbstractSubscription{
 
     synchronized public void add(MessageReference node) throws Exception{
         enqueueCounter++;
-        if(!isFull()&&!isSlaveBroker()){
+        if(!isFull()){
             dispatch(node);
         }else{
             optimizePrefetch();
             synchronized(pending){
-                if( pending.isEmpty() )
-                    if (log.isDebugEnabled()){
+                if( pending.isEmpty() ) {
                     log.debug("Prefetch limit.");
                 }
                 pending.addLast(node);
@@ -74,21 +73,18 @@ abstract public class PrefetchSubscription extends AbstractSubscription{
         }
     }
 
-    public void processMessageDispatchNotification(MessageDispatchNotification mdn){
+    synchronized public void processMessageDispatchNotification(MessageDispatchNotification mdn) throws Exception {
         synchronized(pending){
             for(Iterator i=pending.iterator();i.hasNext();){
                 MessageReference node=(MessageReference) i.next();
                 if(node.getMessageId().equals(mdn.getMessageId())){
                     i.remove();
-                    try{
-                        MessageDispatch md=createMessageDispatch(node,node.getMessage());
+                    createMessageDispatch(node,node.getMessage());
                     dispatched.addLast(node);
-                    }catch(Exception e){
-                        log.error("Problem processing MessageDispatchNotification: "+mdn,e);
-                    }
-                    break;
+                    return;
                 }
             }
+            throw new JMSException("Slave broker out of sync with master: Dispatched message ("+mdn.getMessageId()+") was not in the pending list: "+pending);
         }
     }
 
@@ -178,8 +174,13 @@ abstract public class PrefetchSubscription extends AbstractSubscription{
             }
             throw new JMSException("Could not correlate acknowledgment with dispatched message: "+ack);
         }
+        
+        if( isSlaveBroker() ) {
+        	throw new JMSException("Slave broker out of sync with master: Acknowledgment ("+ack+") was not in the dispatch list: "+dispatched);
+        } else {
         	throw new JMSException("Invalid acknowledgment: "+ack);
         }
+    }
 
     /**
      * @param context
@@ -201,8 +202,12 @@ abstract public class PrefetchSubscription extends AbstractSubscription{
         }
     }
 
+    /**
+     * Used to determine if the broker can dispatch to the consumer.
+     * @return
+     */
     protected boolean isFull(){
-        return dispatched.size()-prefetchExtension>=info.getPrefetchSize();
+        return isSlaveBroker() || dispatched.size()-prefetchExtension>=info.getPrefetchSize();
     }
     
     /**

activemq-core/src/main/java/org/apache/activemq/broker/region/Subscription.java

@@ -95,8 +95,9 @@ public interface Subscription {
     /**
      * Used by a Slave Broker to update dispatch infomation
      * @param mdn
+     * @throws Exception 
      */
-    void processMessageDispatchNotification(MessageDispatchNotification  mdn);
+    void processMessageDispatchNotification(MessageDispatchNotification  mdn) throws Exception;
     
     /**
      * @return true if the broker is currently in slave mode