2 files changed, 49 insertions, 12 deletions

diff --git a/cpp/src/qpid/broker/Queue.cpp b/cpp/src/qpid/broker/Queue.cpp
index 3c8c237b98..08ee133981 100644
--- a/cpp/src/qpid/broker/Queue.cpp
+++ b/cpp/src/qpid/broker/Queue.cpp
@@ -677,21 +677,25 @@ void Queue::setLastNodeFailure()
 {
     if (persistLastNode){
         Mutex::ScopedLock locker(messageLock);
-    	for ( Messages::iterator i = messages.begin(); i != messages.end(); ++i ) {
-            if (lastValueQueue) checkLvqReplace(*i);
-            // don't force a message twice to disk.
-            if(!i->payload->isStoredOnQueue(shared_from_this())) {
-                i->payload->forcePersistent();
-                if (i->payload->isForcedPersistent() ){
-            	    enqueue(0, i->payload);
+        try {
+    	    for ( Messages::iterator i = messages.begin(); i != messages.end(); ++i ) {
+                if (lastValueQueue) checkLvqReplace(*i);
+                // don't force a message twice to disk.
+                if(!i->payload->isStoredOnQueue(shared_from_this())) {
+                    i->payload->forcePersistent();
+                    if (i->payload->isForcedPersistent() ){
+            	        enqueue(0, i->payload);
+                    }
                 }
-            }
-    	}
+    	    }
+        } catch (const std::exception& e) {
+            // Could not go into last node standing (for example journal not large enough)
+            QPID_LOG(error, "Unable to fail to last node standing for queue: " << name << " : " << e.what());
+        }
         inLastNodeFailure = true;
     }
 }
 
-
 // return true if store exists, 
 bool Queue::enqueue(TransactionContext* ctxt, boost::intrusive_ptr<Message> msg)
 {
diff --git a/cpp/src/tests/QueueTest.cpp b/cpp/src/tests/QueueTest.cpp
index be4a87d2ab..b70afa52a7 100644
--- a/cpp/src/tests/QueueTest.cpp
+++ b/cpp/src/tests/QueueTest.cpp
@@ -19,6 +19,7 @@
  *
  */
 #include "unit_test.h"
+#include "test_tools.h"
 #include "qpid/Exception.h"
 #include "qpid/broker/Broker.h"
 #include "qpid/broker/Queue.h"
@@ -275,11 +276,13 @@ class TestMessageStoreOC : public NullMessageStore
 
     uint enqCnt;
     uint deqCnt;
+    bool error;
     
     virtual void dequeue(TransactionContext*,
                  const boost::intrusive_ptr<PersistableMessage>& /*msg*/,
                  const PersistableQueue& /*queue*/)
     {
+        if (error) throw Exception("Dequeue error test");
         deqCnt++;
     }
 
@@ -287,10 +290,16 @@ class TestMessageStoreOC : public NullMessageStore
                  const boost::intrusive_ptr<PersistableMessage>& /*msg*/,
                  const PersistableQueue& /* queue */)
     {
+        if (error) throw Exception("Enqueue error test");
         enqCnt++;
     }
 
-    TestMessageStoreOC() : NullMessageStore(),enqCnt(0),deqCnt(0) {}
+    void createError()
+    {
+        error=true;
+    }
+    
+    TestMessageStoreOC() : NullMessageStore(),enqCnt(0),deqCnt(0),error(false) {}
     ~TestMessageStoreOC(){}
 };
 
@@ -689,6 +698,30 @@ not requeued to the store.
 
 }
 
-QPID_AUTO_TEST_SUITE_END()
+QPID_AUTO_TEST_CASE(testLastNodeJournalError){
+/*
+simulate store excption going into last node standing
+
+*/
+    TestMessageStoreOC  testStore;
+    client::QueueOptions args;
+    // set queue mode
+    args.setPersistLastNode();
+
+    Queue::shared_ptr queue1(new Queue("my-queue", true, &testStore));
+    intrusive_ptr<Message> received;
+    queue1->configure(args);
+ 	
+    // check requeue 1
+    intrusive_ptr<Message> msg1 = create_message("e", "C");
+
+    queue1->deliver(msg1);
+    testStore.createError();
+    
+    ScopedSuppressLogging sl; // Suppress messages for expected errors.
+    queue1->setLastNodeFailure();
+    BOOST_CHECK_EQUAL(testStore.enqCnt, 0u);
+
+}QPID_AUTO_TEST_SUITE_END()