0000926: Staging area is not being cleaned up. Can cause out of memor…

…y or file system issues.
JumpMind · Nov 27, 2012 · aa95b1f · aa95b1f
1 parent bc204dc
commit aa95b1f
Show file tree

Hide file tree

Showing 4 changed files with 35 additions and 10 deletions.
diff --git a/symmetric-client/src/main/java/org/jumpmind/symmetric/job/StageManagementJob.java b/symmetric-client/src/main/java/org/jumpmind/symmetric/job/StageManagementJob.java
@@ -3,6 +3,7 @@
 import org.jumpmind.symmetric.ISymmetricEngine;
 import org.jumpmind.symmetric.io.stage.IStagingManager;
 import org.jumpmind.symmetric.service.ClusterConstants;
+import org.jumpmind.symmetric.service.IOutgoingBatchService;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
 
 public class StageManagementJob extends AbstractJob {
@@ -13,6 +14,7 @@ public StageManagementJob(ISymmetricEngine engine, ThreadPoolTaskScheduler taskS
             IStagingManager stagingManager) {
         super("job.stage.management", true, engine.getParameterService().is(
                 "start.stage.management.job"), engine, taskScheduler);
+        this.stagingManager = stagingManager;
     }
 
     public String getClusterLockName() {
@@ -28,6 +30,9 @@ void doJob(boolean force) throws Exception {
         if (stagingManager != null) {
             stagingManager.clean();
         }
+
+        IOutgoingBatchService outgoingBatchService = engine.getOutgoingBatchService();
+
     }
 
 }
diff --git a/symmetric-io/src/main/java/org/jumpmind/symmetric/io/stage/IStagedResource.java b/symmetric-io/src/main/java/org/jumpmind/symmetric/io/stage/IStagedResource.java
@@ -50,6 +50,8 @@ public String getExtensionName() {
 
     public long getCreateTime();
 
+    public boolean isFileResource();
+
     public void delete();
 
     public boolean exists();

diff --git a/symmetric-io/src/main/java/org/jumpmind/symmetric/io/stage/StagedResource.java b/symmetric-io/src/main/java/org/jumpmind/symmetric/io/stage/StagedResource.java
@@ -90,6 +90,10 @@ public StagedResource(long threshold, File directory, String path, StagingManage
         createTime = System.currentTimeMillis();
         this.state = State.CREATE;
     }
+
+    public boolean isFileResource() {     
+        return file != null && file.exists();
+    }
 
     protected File buildFile(State state) {
         return new File(directory, String.format("%s.%s", path, state.getExtensionName()));

diff --git a/symmetric-io/src/main/java/org/jumpmind/symmetric/io/stage/StagingManager.java b/symmetric-io/src/main/java/org/jumpmind/symmetric/io/stage/StagingManager.java
@@ -81,21 +81,35 @@ protected void refreshResourceList() {
     public long clean() {
         this.refreshResourceList();
         Set<String> keys = new HashSet<String>(resourceList.keySet());
-        long purgedCount = 0;
-        long purgedSize = 0;
+        long purgedFileCount = 0;
+        long purgedFileSize = 0;
+        long purgedMemCount = 0;
+        long purgedMemSize = 0;        
         for (String key : keys) {
             IStagedResource resource = resourceList.get(key);
-            boolean resourceIsOld = System.currentTimeMillis() - resource.getCreateTime() > timeToLiveInMs;
-            if (resource.getState() == State.DONE && (resourceIsOld || !resource.exists())) {
-                purgedCount++;
-                purgedSize += resource.getSize();
+            boolean resourceIsOld = (System.currentTimeMillis() - resource.getCreateTime()) > timeToLiveInMs;
+            if ((resource.getState() == State.READY || resource.getState() == State.DONE) && 
+                (resourceIsOld || !resource.exists())) {
+                if (resource.isFileResource()) {
+                    purgedFileCount++;
+                    purgedFileSize += resource.getSize();
+                } else {
+                    purgedMemCount++;
+                    purgedMemSize += resource.getSize();                    
+                }
                 resource.delete();
                 resourceList.remove(key);
             }
         }
-        log.info("Purged {} staged files, freeing {} kb of disk space", purgedCount,
-                (int) (purgedSize / 1000));
-        return purgedCount;
+        if (purgedFileCount > 0) {
+            log.info("Purged {} staged files, freeing {} kb of disk space", purgedFileCount,
+                    (int) (purgedFileSize / 1000));
+        }
+        if (purgedMemCount > 0) {
+            log.info("Purged {} staged memory buffers, freeing {} kb of memory", purgedMemCount,
+                    (int) (purgedMemSize / 1000));
+        }
+        return purgedFileCount;
     }
 
     /**
@@ -129,7 +143,7 @@ public IStagedResource find(Object... path) {
         IStagedResource resource = resourceList.get(filePath);
         if (resource != null) {
             if (!resource.exists()
-                    && (resource.getState() == State.READY || resource.getState() == State.READY)) {
+                    && (resource.getState() == State.READY || resource.getState() == State.DONE)) {
                 resource.delete();
                 resource = null;
             }