6722: Purge of batches before data causes appearance of stranded data (#226)

0006722: Purge of batches before data causes appearance of stranded data. Prevent NPE

Author: erilong

symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/DataLoaderService.java

@@ -1084,7 +1084,9 @@ protected IDataWriter chooseDataWriter(Batch batch) {
                                 log.info("Bulk loader failed in class {} with message: {}", e.getClass().getName(), e.getMessage());
                                 ctx.put(ContextConstants.CONTEXT_BULK_WRITER_TO_USE, "default");
                                 ctx.setLastError(null);
-                                listener.currentBatch.setStatus(Status.OK);
+                                if (listener.currentBatch != null) {
+                                    listener.currentBatch.setStatus(Status.OK);
+                                }
                                 processor.setDataReader(buildDataReader(batchInStaging, resource));
                                 try {
                                     listener.getBatchesProcessed().remove(listener.currentBatch);
@@ -1099,7 +1101,7 @@ protected IDataWriter chooseDataWriter(Batch batch) {
                                 }
                             } else {
                                 isError = true;
-                                if (listener.currentBatch.getSqlCode() == ErrorConstants.PROTOCOL_VIOLATION_CODE) {
+                                if (listener.currentBatch != null && listener.currentBatch.getSqlCode() == ErrorConstants.PROTOCOL_VIOLATION_CODE) {
                                     log.info("The batch {} may be corrupt in staging, so removing it.", batchInStaging.getNodeBatchId());
                                     resource.delete();
                                     incomingBatch = listener.currentBatch;

symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/PurgeService.java

@@ -29,7 +29,6 @@
 import java.util.List;
 import java.util.function.LongConsumer;
 
-import org.apache.commons.lang3.StringUtils;
 import org.apache.commons.lang3.time.DateUtils;
 import org.apache.commons.lang3.time.FastDateFormat;
 import org.jumpmind.db.platform.DatabaseNamesConstants;
@@ -234,7 +233,7 @@ private long[] getMinMax(long[] minMax, long notOkBatchId, long[] rangeMinMax) {
         if (rangeMinMax[1] == minMax[1]) {
             minMax[1] = -1;
         } else {
-            minMax[0] = notOkBatchId + 1;
+            minMax[0] = notOkBatchId;
         }
         return minMax;
     }
@@ -281,8 +280,8 @@ private long purgeLingeringBatches(OutgoingContext context) {
             } else {
                 break;
             }
-            log.info("Done purging {} lingering batches and {} rows", totalBatchesPurged, totalRowsPurged);
         }
+        log.info("Done purging {} lingering batches and {} rows", totalBatchesPurged, totalRowsPurged);
         return totalRowsPurged;
     }
 
@@ -926,22 +925,25 @@ private OutgoingContext buildOutgoingContext(Calendar retentionCutoff) {
         context.setMinEventBatchId(startEventBatchId);
         // Leave 1 batch and its data around so MySQL auto increment doesn't reset
         long endBatchId = sequenceService.currVal(Constants.SEQUENCE_OUTGOING_BATCH) - 1;
-        List<Long> batchIds = sqlTemplateDirty.query(getSql("maxBatchIdByChannel"), new LongMapper(),
+        List<Long> batchIds = sqlTemplateDirty.query(getSql("maxBatchIdForOldBatches"), new LongMapper(),
                 new Object[] { startBatchId, endBatchId, new Timestamp(context.getRetentionCutoff().getTime().getTime()) },
                 new int[] { symmetricDialect.getSqlTypeForIds(), symmetricDialect.getSqlTypeForIds(), Types.TIMESTAMP });
         if (batchIds != null && batchIds.size() > 0) {
-            int[] types = new int[batchIds.size()];
-            for (int i = 0; i < batchIds.size(); i++) {
-                types[i] = symmetricDialect.getSqlTypeForIds();
-                if (batchIds.get(i) > context.getMaxBatchId()) {
-                    context.setMaxBatchId(batchIds.get(i));
+            context.setMaxBatchId(batchIds.get(0));
+            log.info("Max eligible batch ID: {}", context.getMaxBatchId());
+            List<Row> rows = sqlTemplateDirty.query(getSql("minMaxDataIdForOldBatches"), new Object[] { startBatchId,
+                    context.getMaxBatchId() }, new int[] { symmetricDialect.getSqlTypeForIds(), symmetricDialect.getSqlTypeForIds() });
+            if (rows != null && rows.size() > 0) {
+                Row row = rows.get(0);
+                long minDataId = row.getLong("min_data_id");
+                long maxDataId = row.getLong("max_data_id");
+                context.setMaxDataId(maxDataId);
+                log.info("Max eligible data ID: {}", context.getMaxDataId());
+                if (minDataId < context.getMinDataId()) {
+                    log.info("Moving starting data ID back from {} to {}", context.getMinDataId(), minDataId);
+                    context.setMinDataId(minDataId);
                 }
             }
-            String sql = getSql("maxDataIdForBatches").replace("?", StringUtils.repeat("?", ",", batchIds.size()));
-            List<Long> ids = sqlTemplateDirty.query(sql, new LongMapper(), batchIds.toArray(), types);
-            if (ids != null && ids.size() > 0) {
-                context.setMaxDataId(ids.get(0));
-            }
         }
         context.setMinDataGapStartId(sqlTemplateDirty.queryForLong(getSql("minDataGapStartId")));
         context.setDataGapsExpired(dataService.findDataGapsExpired());

symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/PurgeServiceSqlMap.java

@@ -67,9 +67,9 @@ public PurgeServiceSqlMap(IDatabasePlatform platform, Map<String, String> replac
 
         putSql("minDataEventId", "select min(data_id) from $(data_event)");
 
-        putSql("maxBatchIdByChannel", "select max(batch_id) from $(outgoing_batch) where batch_id between ? and ? and create_time < ? group by channel_id");
+        putSql("maxBatchIdForOldBatches", "select max(batch_id) from $(outgoing_batch) where batch_id between ? and ? and create_time < ?");
 
-        putSql("maxDataIdForBatches", "select max(data_id) from $(data_event) where batch_id in (?)");
+        putSql("minMaxDataIdForOldBatches", "select min(data_id) as min_data_id, max(data_id) as max_data_id from $(data_event) where batch_id between ? and ?");
 
         putSql("selectNodesWithStrandedBatches", "select distinct node_id from $(outgoing_batch) " + 
 "where node_id not in (select node_id from $(node) where sync_enabled = ?) and status != ?");