0003183: After a failed extract in background batches should be cleaned

up and the batch status should be set back to RQ
JumpMind · Jul 3, 2017 · 5c83e49 · 5c83e49
1 parent e5148e2
commit 5c83e49
Showing 1 changed file with 14 additions and 3 deletions.
diff --git a/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/DataExtractorService.java b/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/DataExtractorService.java
@@ -1506,10 +1506,10 @@ public void execute(NodeCommunication nodeCommunication, RemoteNodeStatus status
                         resource.delete();
                     }
 
-                    MultiBatchStagingWriter multiBatchStatingWriter = 
+                    MultiBatchStagingWriter multiBatchStagingWriter = 
                             buildMultiBatchStagingWriter(request, identity, targetNode, batches, processInfo, channel);
 
-                    extractOutgoingBatch(processInfo, targetNode, multiBatchStatingWriter, 
+                    extractOutgoingBatch(processInfo, targetNode, multiBatchStagingWriter, 
                             firstBatch, false, false, ExtractMode.FOR_SYM_CLIENT);
 
                     for (OutgoingBatch outgoingBatch : batches) {
@@ -1590,10 +1590,21 @@ public void execute(NodeCommunication nodeCommunication, RemoteNodeStatus status
                         request.getEndBatchId() });
                 processInfo.setStatus(org.jumpmind.symmetric.model.ProcessInfo.Status.OK);
             } catch (RuntimeException ex) {
-                log.debug(
+                log.warn(
                         "Failed to extract batches for request {}. Starting at batch {}.  Ending at batch {}",
                         new Object[] { request.getRequestId(), request.getStartBatchId(),
                                 request.getEndBatchId() });
+                List<OutgoingBatch> checkBatches = outgoingBatchService.getOutgoingBatchRange(
+                        request.getStartBatchId(), request.getEndBatchId()).getBatches();
+                for (OutgoingBatch outgoingBatch : checkBatches) {
+                    outgoingBatch.setStatus(Status.RQ);
+                    IStagedResource resource = getStagedResource(outgoingBatch);
+                    if (resource != null) {
+                        resource.close();
+                        resource.delete();
+                    }
+                    outgoingBatchService.updateOutgoingBatch(outgoingBatch);
+                }                
                 processInfo.setStatus(org.jumpmind.symmetric.model.ProcessInfo.Status.ERROR);
                 throw ex;
             }