QPID-949 : Removed the deleteOnExit calls on all flowed entries, as this will increase our memory usage. Instead add a close method on the BackingStore that is called when the queue closes and clean up the backing then. Also moved the QueueHousekeeping thread stop to before we do any queue closing in the VHost. This will ensure that we are not causing any operations that might inadvertadly load a message, and so prevent the backing file from being deleted. This should of course now not occur as all getMessage() calls have been removed.

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

5 files changed, 24 insertions, 14 deletions

diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/FileQueueBackingStore.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/FileQueueBackingStore.java
index 7c83788883..0e5a4efba6 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/FileQueueBackingStore.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/FileQueueBackingStore.java
@@ -28,6 +28,7 @@ import org.apache.qpid.framing.BasicContentHeaderProperties;
 import org.apache.qpid.framing.ContentHeaderBody;
 import org.apache.qpid.framing.abstraction.ContentChunk;
 import org.apache.qpid.framing.abstraction.MessagePublishInfo;
+import org.apache.qpid.util.FileUtils;
 
 import java.io.File;
 import java.io.FileInputStream;
@@ -55,7 +56,6 @@ public class FileQueueBackingStore implements QueueBackingStore
         MessageMetaData mmd;
 
         File handle = getFileHandle(messageId);
-        handle.deleteOnExit();
 
         ObjectInputStream input = null;
 
@@ -192,8 +192,6 @@ public class FileQueueBackingStore implements QueueBackingStore
             _log.info("Unloading Message (ID:" + messageId + ")");
         }
 
-        handle.deleteOnExit();
-
         ObjectOutputStream writer = null;
         Exception error = null;
 
@@ -295,7 +293,6 @@ public class FileQueueBackingStore implements QueueBackingStore
         if (!bin_dir.exists())
         {
             bin_dir.mkdirs();
-            bin_dir.deleteOnExit();
         }
 
         String id = bin_path + File.separator + messageId;
@@ -304,7 +301,7 @@ public class FileQueueBackingStore implements QueueBackingStore
     }
 
     public void delete(Long messageId)
-    {        
+    {
         File handle = getFileHandle(messageId);
 
         if (handle.exists())
@@ -320,6 +317,15 @@ public class FileQueueBackingStore implements QueueBackingStore
         }
     }
 
+    public void close()
+    {
+        _log.info("Closing Backing store at:" + _flowToDiskLocation);
+        if (!FileUtils.delete(new File(_flowToDiskLocation), true))
+        {
+            _log.error("Unable to fully delete backing store location");
+        }
+    }
+
     private class RecoverDataBuffer implements ContentChunk
     {
         private int _length;
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/FileQueueBackingStoreFactory.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/FileQueueBackingStoreFactory.java
index 0cfa9d6b32..21073c22ae 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/FileQueueBackingStoreFactory.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/FileQueueBackingStoreFactory.java
@@ -56,6 +56,7 @@ public class FileQueueBackingStoreFactory implements QueueBackingStoreFactory
 
         _flowToDiskLocation += File.separator + QUEUE_BACKING_DIR + File.separator + vHostName;
 
+        //Check the location we will create QUEUE_BACKING_DIR in.
         File root = new File(location);
         if (!root.exists())
         {
@@ -121,8 +122,7 @@ public class FileQueueBackingStoreFactory implements QueueBackingStoreFactory
 
         _log.info("Creating Flow to Disk Store : " + store.getAbsolutePath());
         store.deleteOnExit();
-
-        if(!store.mkdir())
+        if (!store.mkdir())
         {
             throw new ConfigurationException("Unable to create Temporary Flow to Disk store:" + store.getAbsolutePath());
         }
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/FlowableBaseQueueEntryList.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/FlowableBaseQueueEntryList.java
index d25c096337..0c4b8a0b42 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/FlowableBaseQueueEntryList.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/FlowableBaseQueueEntryList.java
@@ -236,6 +236,8 @@ public abstract class FlowableBaseQueueEntryList implements QueueEntryList
             //Shutdown thread for inhaler.
             ReferenceCountingExecutorService.getInstance().releaseExecutorService();
             ReferenceCountingExecutorService.getInstance().releaseExecutorService();
+
+            _backingStore.close();
         }
     }
 
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/QueueBackingStore.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/QueueBackingStore.java
index 1f575d1e05..5efb95d0c0 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/QueueBackingStore.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/QueueBackingStore.java
@@ -31,4 +31,6 @@ public interface QueueBackingStore
     void unload(AMQMessage message) throws UnableToFlowMessageException;
 
     void delete(Long messageId);
+
+    void close();
 }
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java
index db05c7b299..5d2a31b80d 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java
@@ -422,6 +422,12 @@ public class VirtualHost implements Accessable
         //Stop Connections
         _connectionRegistry.close();
 
+        //Stop Housekeeping
+        if (_houseKeepingTimer != null)
+        {
+            _houseKeepingTimer.cancel();
+        }
+
         //Stop the Queues processing
         if (_queueRegistry != null)
         {
@@ -429,13 +435,7 @@ public class VirtualHost implements Accessable
             {
                 queue.stop();
             }
-        }        
-
-        //Stop Housekeeping
-        if (_houseKeepingTimer != null)
-        {
-            _houseKeepingTimer.cancel();
-        }        
+        }
 
         //Close TransactionLog
         if (_transactionLog != null)