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
index 779ed7781c..e7d77c5334 100644
--- 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
@@ -1049,7 +1049,16 @@ protected IDataWriter chooseDataWriter(Batch batch) {
                                 ctx.put(ContextConstants.CONTEXT_BULK_WRITER_TO_USE, "default");
                                 listener.currentBatch.setStatus(Status.OK);
                                 processor.setDataReader(buildDataReader(batchInStaging, resource));
-                                processor.process(ctx);
+                                try {
+                                    listener.getBatchesProcessed().remove(listener.currentBatch);
+                                    processor.process(ctx);
+                                } catch (Exception retryException) {
+                                    isError = true;
+                                    incomingBatch = listener.currentBatch;
+                                    incomingBatch.setStatus(Status.ER);
+                                    incomingBatchService.updateIncomingBatch(incomingBatch);
+                                    throw e;
+                                }
                             } else {
                                 if (e instanceof ParseException || e instanceof ProtocolException) {
                                     log.info("The batch {} may be corrupt in staging, so removing it.", batchInStaging.getNodeBatchId());