0005288: Made adjustments to make it possible to ignore a row from an…

… outgoing batch that is part of a load from the Pro UI
JumpMind · May 4, 2022 · 2ac5a38 · 2ac5a38
1 parent 5b4860d
commit 2ac5a38
Show file tree

Hide file tree

Showing 5 changed files with 19 additions and 2 deletions.
diff --git a/symmetric-core/src/main/java/org/jumpmind/symmetric/service/IDataExtractorService.java b/symmetric-core/src/main/java/org/jumpmind/symmetric/service/IDataExtractorService.java
@@ -26,6 +26,7 @@
 
 import org.jumpmind.db.sql.ISqlTransaction;
 import org.jumpmind.symmetric.io.data.writer.StructureDataWriter.PayloadType;
+import org.jumpmind.symmetric.io.stage.StagingFileLock;
 import org.jumpmind.symmetric.model.ExtractRequest;
 import org.jumpmind.symmetric.model.Node;
 import org.jumpmind.symmetric.model.OutgoingBatch;
@@ -67,6 +68,8 @@ public ExtractRequest requestExtractRequest(ISqlTransaction transaction, String
     public void resetExtractRequest(OutgoingBatch batch);
 
     public void removeBatchFromStaging(OutgoingBatch batch);
+
+    public StagingFileLock acquireStagingFileLock(OutgoingBatch batch);
 
     public List<ExtractRequest> getPendingTablesForExtractByLoadId(long loadId);
 

diff --git a/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/AcknowledgeService.java b/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/AcknowledgeService.java
@@ -120,8 +120,8 @@ public BatchAckResult ack(final BatchAck batch) {
                             isNewError = outgoingBatch.getFailedDataId() == 0 || outgoingBatch.getFailedDataId() != failedDataId;
                             outgoingBatch.setFailedDataId(failedDataId);
                         }
-                        outgoingBatch.setFailedLineNumber(batch.getErrorLine());
                     }
+                    outgoingBatch.setFailedLineNumber(batch.getErrorLine());
                 }
                 if (status == Status.ER) {
                     boolean suppressError = false;

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
@@ -998,7 +998,7 @@ private BatchLock acquireLock(OutgoingBatch batch, boolean useStagingDataWriter)
         return lock;
     }
 
-    protected StagingFileLock acquireStagingFileLock(OutgoingBatch batch) {
+    public StagingFileLock acquireStagingFileLock(OutgoingBatch batch) {
         boolean stagingFileAcquired = false;
         StagingFileLock fileLock = null;
         int iterations = 0;

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
@@ -46,6 +46,8 @@ public String getExtensionName() {
     public File getFile();
 
     public void close();
+
+    public void closeReaders();
 
     public long getSize();
 

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
@@ -298,6 +298,18 @@ private void closeInternal() {
             closeInputStreamsMap();
         }
     }
+
+    public void closeReaders() {
+        if (readers != null) {
+            for (BufferedReader reader : readers.values()) {
+                try {
+                    reader.close();
+                } catch (IOException e) {
+                }
+            }
+            readers = null;
+        }
+    }
 
     public OutputStream getOutputStream() {
         refreshLastUpdateTime();