CDM-29 default behaviour no longer autocorrects counters unless explicitly enabled

Author: mieslep

SIT/smoke/04_counters/breakData.cql

@@ -2,6 +2,7 @@
 UPDATE source.smoke_counters SET col1=col1+1 WHERE key='record2';
 
 -- CDM-29 discovered that this record will "zombie"...
+-- fixData will not correct it, fixForce will
 DELETE FROM target.smoke_counters where key='record3';
 
 -- change by same amount on both sides, records should match

SIT/smoke/04_counters/cdm.fixData.assert

@@ -2,6 +2,6 @@ Read Record Count: 5
 Mismatch Record Count: 2
 Corrected Mismatch Record Count: 2
 Missing Record Count: 1
-Corrected Missing Record Count: 1
+Corrected Missing Record Count: 0
 Valid Record Count: 2
 Skipped Record Count: 0

SIT/smoke/04_counters/cdm.fixForce.assert

@@ -0,0 +1,7 @@
+Read Record Count: 5
+Mismatch Record Count: 0
+Corrected Mismatch Record Count: 0
+Missing Record Count: 1
+Corrected Missing Record Count: 1
+Valid Record Count: 4
+Skipped Record Count: 0

SIT/smoke/04_counters/cdm.txt

@@ -1,3 +1,4 @@
 migrateData datastax.astra.migrate.Migrate migrate.properties
 validateData datastax.astra.migrate.DiffData migrate.properties
 fixData datastax.astra.migrate.DiffData fix.properties
+fixForce datastax.astra.migrate.DiffData fix-force.properties

SIT/smoke/04_counters/execute.sh

@@ -13,3 +13,6 @@ cqlsh -u $CASS_USERNAME -p $CASS_PASSWORD $CASS_CLUSTER -f $workingDir/breakData
 
 /local/cdm.sh -f cdm.txt -s fixData -d "$workingDir" > cdm.fixData.out 2>cdm.fixData.err
 /local/cdm-assert.sh -f cdm.fixData.out -a cdm.fixData.assert -d "$workingDir"
+
+/local/cdm.sh -f cdm.txt -s fixForce -d "$workingDir" > cdm.fixForce.out 2>cdm.fixForce.err
+/local/cdm-assert.sh -f cdm.fixForce.out -a cdm.fixForce.assert -d "$workingDir"

SIT/smoke/04_counters/fix-force.properties

@@ -0,0 +1,24 @@
+spark.origin.host                                 cdm-sit-cass
+spark.origin.port                                 9042
+spark.origin.username                             cassandra
+spark.origin.password                             cassandra
+spark.origin.keyspaceTable                        source.smoke_counters
+
+spark.target.host                                 cdm-sit-cass
+spark.target.port                                 9042
+spark.target.username                             cassandra
+spark.target.password                             cassandra
+spark.target.keyspaceTable                        target.smoke_counters
+
+spark.numSplits                                   1
+
+spark.query.origin                                key,col1,col2
+spark.query.types                                 0,2,2
+spark.query.origin.partitionKey                   key
+spark.query.target.id                             key
+
+spark.counterTable.cql.index                      1,2
+spark.counterTable.missing.force                  true
+
+spark.target.autocorrect.missing                  true
+spark.target.autocorrect.mismatch                 true

src/main/java/datastax/astra/migrate/DiffJobSession.java

@@ -41,6 +41,7 @@ public class DiffJobSession extends CopyJobSession {
     private final AtomicLong skippedCounter = new AtomicLong(0);
 
     private final boolean isCounterTable;
+    private final boolean forceCounterWhenMissing;
     private final List<Integer> targetToOriginColumnIndexes;
     private final List<MigrateDataType> targetColumnTypes;
     private final int explodeMapKeyIndex;
@@ -56,6 +57,7 @@ private DiffJobSession(CqlSession originSession, CqlSession targetSession, Spark
         logger.info("PARAM -- Autocorrect Mismatch: {}", autoCorrectMismatch);
 
         this.isCounterTable = cqlHelper.isCounterTable();
+        this.forceCounterWhenMissing = propertyHelper.getBoolean(KnownProperties.ORIGIN_COUNTER_FORCE_WHEN_MISSING);
         this.targetToOriginColumnIndexes = cqlHelper.getPKFactory().getTargetToOriginColumnIndexes();
         this.targetColumnTypes = cqlHelper.getPKFactory().getTargetColumnTypes();
 
@@ -167,6 +169,10 @@ private void diff(Record record) {
         if (targetRow == null) {
             missingCounter.incrementAndGet();
             logger.error("Missing target row found for key: {}", record.getPk());
+            if (autoCorrectMissing && isCounterTable && !forceCounterWhenMissing) {
+                logger.error("{} is true, but not Inserting as {} is not enabled; key : {}", KnownProperties.TARGET_AUTOCORRECT_MISSING, KnownProperties.ORIGIN_COUNTER_FORCE_WHEN_MISSING, record.getPk());
+                return;
+            }
 
             //correct data
             if (autoCorrectMissing) {

src/main/java/datastax/astra/migrate/properties/KnownProperties.java

@@ -71,6 +71,7 @@ public enum PropertyType {
     public static final String ORIGIN_IS_COUNTER      = "spark.counterTable";              // false
     public static final String ORIGIN_COUNTER_CQL     = "spark.counterTable.cql";
     public static final String ORIGIN_COUNTER_INDEXES = "spark.counterTable.cql.index";    // 0
+    public static final String ORIGIN_COUNTER_FORCE_WHEN_MISSING = "spark.counterTable.missing.force";    // false
     public static final String ORIGIN_PRIMARY_KEY        = "spark.query.origin.id";         // Defaults to TARGET_PRIMARY_KEY, same config rules
     public static final String ORIGIN_PRIMARY_KEY_TYPES  = "spark.cdm.cql.origin.id.types"; // Code-managed, not an external property
 
@@ -89,6 +90,8 @@ public enum PropertyType {
         defaults.put(ORIGIN_IS_COUNTER, "false");
            types.put(ORIGIN_COUNTER_CQL, PropertyType.STRING);
            types.put(ORIGIN_COUNTER_INDEXES, PropertyType.NUMBER_LIST);
+           types.put(ORIGIN_COUNTER_FORCE_WHEN_MISSING, PropertyType.BOOLEAN);
+        defaults.put(ORIGIN_COUNTER_FORCE_WHEN_MISSING, "false");
            types.put(ORIGIN_PRIMARY_KEY, PropertyType.STRING_LIST);
         required.add(ORIGIN_PRIMARY_KEY);
            types.put(ORIGIN_PRIMARY_KEY_TYPES, PropertyType.MIGRATION_TYPE_LIST);