QPID-1793 : Update AMQChannel to remember any AMQException that occurs during requeue and then throws that after it has attempted to requeue all the messsagse from the unackedMap.

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

1 files changed, 28 insertions, 9 deletions

diff --git a/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java b/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
index dcbd887896..cb19532872 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
@@ -490,26 +490,45 @@ public class AMQChannel
             }
         }
 
+        // Place to hold any error that occured during the requeueing.
+        AMQException error = null;
         for (QueueEntry unacked : messagesToBeDelivered)
         {
-            if (!unacked.isQueueDeleted())
+            try
             {
-                // Mark message redelivered
-                unacked.setRedelivered(true);
+                if (!unacked.isQueueDeleted())
+                {
+                    // Mark message redelivered
+                    unacked.setRedelivered(true);
 
-                // Ensure message is released for redelivery
-                unacked.release();
+                    // Ensure message is released for redelivery
+                    unacked.release();
 
-                // Deliver Message
-                deliveryContext.requeue(unacked);
+                    // Deliver Message
+                    deliveryContext.requeue(unacked);
 
+                }
+                else
+                {
+                    unacked.dequeueAndDelete(_storeContext);
+                }
             }
-            else
+            catch (AMQException e)
             {
-                unacked.dequeueAndDelete(_storeContext);
+                //Log the error and store it
+                _log.error(e.getMessage(),e);
+                // We store the last seen exception for rethrowing after
+                // attempting to process all the entries.
+                error = e;
             }
         }
 
+        // If we had an error during the requeue process throw it now.
+        if (error != null)
+        {
+            throw error;
+        }
+
     }
 
     /**