0003157: Allow bulk loaders to fall back to default loader when an error

occurs
JumpMind · Jun 13, 2017 · c7d814e · c7d814e
1 parent 561f299
commit c7d814e
Show file tree

Hide file tree

Showing 3 changed files with 20 additions and 1 deletion.
diff --git a/symmetric-client/src/main/java/org/jumpmind/symmetric/io/PostgresBulkDatabaseWriter.java b/symmetric-client/src/main/java/org/jumpmind/symmetric/io/PostgresBulkDatabaseWriter.java
@@ -41,6 +41,7 @@
 import org.jumpmind.symmetric.io.data.writer.DataWriterStatisticConstants;
 import org.jumpmind.symmetric.io.data.writer.DatabaseWriterSettings;
 import org.jumpmind.symmetric.io.data.writer.DefaultDatabaseWriter;
+import org.jumpmind.symmetric.model.IncomingBatch;
 import org.postgresql.copy.CopyIn;
 import org.postgresql.copy.CopyManager;
 import org.postgresql.core.BaseConnection;
@@ -60,6 +61,8 @@ public class PostgresBulkDatabaseWriter extends DefaultDatabaseWriter {
 
     protected boolean needsBinaryConversion;
 
+    protected boolean useDefaultDataWriter;
+
     public PostgresBulkDatabaseWriter(IDatabasePlatform platform, DatabaseWriterSettings settings,
             NativeJdbcExtractor jdbcExtractor, int maxRowsBeforeFlush) {
         super(platform, settings);
@@ -68,6 +71,11 @@ public PostgresBulkDatabaseWriter(IDatabasePlatform platform, DatabaseWriterSett
     }
 
     public void write(CsvData data) {
+        if (useDefaultDataWriter) {
+            super.write(data);
+            return;
+        }
+
         statistics.get(batch).increment(DataWriterStatisticConstants.STATEMENTCOUNT);
         statistics.get(batch).increment(DataWriterStatisticConstants.LINENUMBER);
         statistics.get(batch).startTimer(DataWriterStatisticConstants.DATABASEMILLIS);
@@ -182,6 +190,13 @@ protected void endCopy() {
         }
     }
 
+    @Override
+    public void start(Batch batch) {
+        super.start(batch);
+        IncomingBatch currentBatch = (IncomingBatch) context.get("currentBatch");
+        useDefaultDataWriter = currentBatch == null ? false : currentBatch.isErrorFlag();
+    }
+
     @Override
     public boolean start(Table table) {
         return super.start(table);

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
@@ -964,7 +964,6 @@ public Table nextTable() {
                                     return table;
                                 }                                
                             };
-
                             DataProcessor processor = new DataProcessor(reader, null, listener, "data load from stage") {
                                 @Override
                                 protected IDataWriter chooseDataWriter(Batch batch) {
@@ -1043,6 +1042,8 @@ public void beforeBatchEnd(DataContext context) {
         public boolean beforeBatchStarted(DataContext context) {
             this.currentBatch = null;
             Batch batch = context.getBatch();
+            context.remove("currentBatch");
+
             if (parameterService.is(ParameterConstants.DATA_LOADER_ENABLED)
                     || (batch.getChannelId() != null && batch.getChannelId().equals(
                             Constants.CHANNEL_CONFIG))) {
@@ -1059,6 +1060,8 @@ public boolean beforeBatchStarted(DataContext context) {
                 this.batchesProcessed.add(incomingBatch);
                 if (incomingBatchService.acquireIncomingBatch(incomingBatch)) {
                     this.currentBatch = incomingBatch;
+                    context.put("currentBatch", this.currentBatch);
+
                     return true;
                 }
             }

diff --git a/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/IncomingBatchService.java b/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/IncomingBatchService.java
@@ -214,6 +214,7 @@ public boolean acquireIncomingBatch(IncomingBatch batch) {
                         || !parameterService
                                 .is(ParameterConstants.INCOMING_BATCH_SKIP_DUPLICATE_BATCHES_ENABLED)) {
                     okayToProcess = true;
+                    batch.setErrorFlag(existingBatch.isErrorFlag());
                     existingBatch.setStatus(Status.LD);
                     log.info("Retrying batch {}", batch.getNodeBatchId());
                 } else if (existingBatch.getStatus() == Status.IG) {