0003748: Batch was not complete

JumpMind · Jan 11, 2019 · 7348dc2 · 7348dc2
1 parent 662c068
commit 7348dc2
Show file tree

Hide file tree

Showing 2 changed files with 13 additions and 9 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
@@ -848,7 +848,7 @@ protected List<OutgoingBatch> extract(final ProcessInfo extractInfo, final Node
                             } else if (e instanceof StagingLowFreeSpace) {
                                 log.error("Extract is disabled because disk is almost full: {}", e.getMessage());
                             } else if (e.getCause() instanceof ZipException) {
-                                log.info("The batch {} appears corrupt in staging, so removing it. ({})", currentBatch.getNodeBatchId(), e.getMessage());
+                                log.warn("The batch {} appears corrupt in staging, so removing it. ({})", currentBatch.getNodeBatchId(), e.getMessage());
                                 IStagedResource extractedBatch = getStagedResource(currentBatch);
                                 if (extractedBatch != null) {
                                     extractedBatch.delete();

diff --git a/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/DataLoaderService.java b/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/DataLoaderService.java
@@ -51,6 +51,7 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
+import java.util.zip.ZipException;
 
 import org.apache.commons.lang.StringUtils;
 import org.jumpmind.db.model.Table;
@@ -1096,17 +1097,20 @@ protected IDataWriter chooseDataWriter(Batch batch) {
                                     throw e;
                                 }
                             } else {
-                                if (e instanceof ParseException || e instanceof ProtocolException) {
-                                    log.info("The batch {} may be corrupt in staging, so removing it.", batchInStaging.getNodeBatchId());
+                                isError = true;
+                                if (e instanceof ParseException || e instanceof ProtocolException || e.getCause() instanceof ZipException) {
+                                    log.warn("The batch {} may be corrupt in staging, so removing it.", batchInStaging.getNodeBatchId());
                                     resource.delete();
                                     incomingBatch = listener.currentBatch;
-                                    incomingBatch.setStatus(Status.ER);
-                                    incomingBatch.setSqlCode(ErrorConstants.PROTOCOL_VIOLATION_CODE);
-                                    incomingBatch.setSqlState(ErrorConstants.PROTOCOL_VIOLATION_STATE);
-                                    incomingBatchService.updateIncomingBatch(incomingBatch);
+                                    if (incomingBatch != null) {
+                                        incomingBatch.setStatus(Status.ER);
+                                        incomingBatch.setSqlCode(ErrorConstants.PROTOCOL_VIOLATION_CODE);
+                                        incomingBatch.setSqlState(ErrorConstants.PROTOCOL_VIOLATION_STATE);
+                                        incomingBatchService.updateIncomingBatch(incomingBatch);
+                                    }
+                                } else {
+                                    throw e;
                                 }
-                                isError = true;
-                                throw e;
                             }
                         } finally {
                             incomingBatch = listener.currentBatch;