Merge branch 'main' into CDM-84

Author: msmygit

.github/workflows/snyk-cli-scan.yml

@@ -11,4 +11,4 @@ on:
   workflow_dispatch:
 
 env:
-  SNYK_SEVERITY_THRESHOLD_LEVEL: high
+  SNYK_SEVERITY_THRESHOLD_LEVEL: critical

PERF/cdm-v4.properties

@@ -127,12 +127,15 @@ spark.cdm.schema.target.keyspaceTable             devices.sensor_data
 #                           5323. The corresponding counter in Origin is also 5323. At some point, the Target
 #                           counter gets DELETEd. Should the .missing record be re-inserted before
 #                           the DELETE gets tombstoned, the counter will zombie back to life, and the
-#                           counter will become 5323+5323 = 10646. 
+#                           counter will become 5323+5323 = 10646.
+# spark.tokenRange
+#   .exceptionDir         : When set partition ranges corresponding to the exceptions will be logged in
+#                           file with the name <keyspace>.<table_name> in this directory.
 #-----------------------------------------------------------------------------------------------------------
 spark.cdm.autocorrect.missing                     false
 spark.cdm.autocorrect.mismatch                    false
 #spark.cdm.autocorrect.missing.counter             false
-
+#spark.tokenRange.exceptionDir                      aaa/bbb
 #===========================================================================================================
 # Performance and Operations Parameters affecting throughput, error handling, and similar concerns.
 # 

README.md

@@ -142,6 +142,7 @@ When running in above mode the tool assumes a `partitions.csv` file to be presen
 - Supports migration/validation from and to [Azure Cosmos Cassandra](https://learn.microsoft.com/en-us/azure/cosmos-db/cassandra)
 - Validate migration accuracy and performance using a smaller randomized data-set
 - Supports adding custom fixed `writetime`
+- Validation - Log partitions range level exceptions, use the exceptions file as input for rerun  
 
 # Known Limitations
 - This tool does not migrate `ttl` & `writetime` at the field-level (for optimization reasons). It instead finds the field with the highest `ttl` & the field with the highest `writetime` within an `origin` row and uses those values on the entire `target` row.

SIT/smoke/01_basic_kvp/cdm.sh

@@ -10,7 +10,7 @@ EOF
 spark-submit \
   --properties-file /smoke/01_basic_kvp/migrate.properties \
   --master "local[*]" \
-  --class datastax.astra.migrate.Migrate /local/cassandra-data-migrator.jar 
+  --class datastax.cdm.job.Migrate /local/cassandra-data-migrator.jar
 
 cat <<EOF
 !!!!!!!!
@@ -21,4 +21,4 @@ EOF
 spark-submit \
   --properties-file /smoke/01_basic_kvp/migrate.properties \
   --master "local[*]" \
-  --class datastax.astra.migrate.DiffData /local/cassandra-data-migrator.jar
+  --class datastax.cdm.job.DiffData /local/cassandra-data-migrator.jar

src/main/java/com/datastax/cdm/job/AbstractJobSession.java

@@ -46,7 +46,12 @@ protected AbstractJobSession(CqlSession originSession, CqlSession targetSession,
         rateLimiterTarget = RateLimiter.create(propertyHelper.getInteger(KnownProperties.PERF_RATELIMIT_TARGET));
         maxRetries = propertyHelper.getInteger(KnownProperties.MAX_RETRIES);
 
+        tokenRangeExceptionDir = propertyHelper.getString(KnownProperties.TOKEN_RANGE_EXCEPTION_DIR);
+        exceptionFileName = propertyHelper.getString(KnownProperties.ORIGIN_KEYSPACE_TABLE);
+
         logger.info("PARAM -- Max Retries: {}", maxRetries);
+        logger.info("PARAM -- Token range exception dir: {}", tokenRangeExceptionDir);
+        logger.info("PARAM -- Token range exception file name: {}", exceptionFileName);
         logger.info("PARAM -- Origin Rate Limit: {}", rateLimiterOrigin.getRate());
         logger.info("PARAM -- Target Rate Limit: {}", rateLimiterTarget.getRate());
 

src/main/java/com/datastax/cdm/job/BaseJobSession.java

@@ -30,6 +30,9 @@ public abstract class BaseJobSession {
 
     protected Integer printStatsAfter = 100000;
 
+    protected String tokenRangeExceptionDir;
+    protected String exceptionFileName;
+
     protected BaseJobSession(SparkConf sc) {
         propertyHelper.initializeSparkConf(sc);
         this.featureMap = calcFeatureMap(propertyHelper);

src/main/java/com/datastax/cdm/job/DiffJobSession.java

@@ -13,6 +13,7 @@
 import com.datastax.cdm.cql.statement.OriginSelectByPartitionRangeStatement;
 import com.datastax.cdm.cql.statement.TargetSelectByPKStatement;
 import com.datastax.cdm.properties.KnownProperties;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.logging.log4j.ThreadContext;
 import org.apache.spark.SparkConf;
 import org.slf4j.Logger;
@@ -154,10 +155,20 @@ public void getDataAndDiff(BigInteger min, BigInteger max) {
                 logger.error("Error occurred during Attempt#: {}", attempts, e);
                 logger.error("Error with PartitionRange -- ThreadID: {} Processing min: {} max: {} -- Attempt# {}",
                         Thread.currentThread().getId(), min, max, attempts);
+                if (StringUtils.isNotBlank(tokenRangeExceptionDir) && attempts == maxAttempts) {
+                    logFailedPartitionsInFile(min, max);
+                }
             }
         }
     }
 
+    private void logFailedPartitionsInFile(BigInteger min, BigInteger max) {
+        try {
+            ExceptionHandler.FileAppend(tokenRangeExceptionDir, exceptionFileName, min + "," + max);
+        } catch (Exception ee) {
+            logger.error("Error occurred while writing to token range file min: {} max: {}", min, max, ee);
+        }
+    }
     private void diffAndClear(List<Record> recordsToDiff) {
         for (Record record : recordsToDiff) {
             try {

src/main/java/com/datastax/cdm/job/ExceptionHandler.java

@@ -0,0 +1,32 @@
+package com.datastax.cdm.job;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.StandardOpenOption;
+
+public class ExceptionHandler {
+
+    private static final String NEW_LINE = System.lineSeparator();
+
+    private static void appendToFile(Path path, String content)
+            throws IOException {
+        // if file not exists, create and write to it
+        // otherwise append to the end of the file
+        Files.write(path, content.getBytes(StandardCharsets.UTF_8),
+                StandardOpenOption.CREATE,
+                StandardOpenOption.APPEND);
+    }
+
+    public static void FileAppend(String dir, String fileName, String content) throws IOException {
+
+        //create directory if not already existing
+        Files.createDirectories(Paths.get(dir));
+        Path path = Paths.get(dir + "/" + fileName);
+        appendToFile(path, content + NEW_LINE);
+
+    }
+
+}

src/main/java/com/datastax/cdm/job/SplitPartitions.java

@@ -5,6 +5,8 @@
 
 import java.io.*;
 import java.math.BigInteger;
+import java.nio.file.Files;
+import java.nio.file.Paths;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;

src/main/java/com/datastax/cdm/properties/KnownProperties.java

@@ -137,6 +137,15 @@ public enum PropertyType {
         defaults.put(MAX_RETRIES, "0");
     }
 
+    //==========================================================================
+    // Error handling
+    //==========================================================================
+    public static final String TOKEN_RANGE_EXCEPTION_DIR = "spark.tokenRange.exceptionDir"; // aaa/bbb
+    public static final String PARTITIONS_TOKEN_RANGE_FILE = "spark.input.partitionFile"; // aaa/bbb/filename
+    static {
+            types.put(TOKEN_RANGE_EXCEPTION_DIR, PropertyType.STRING);
+            types.put(PARTITIONS_TOKEN_RANGE_FILE, PropertyType.STRING);
+    }
     //==========================================================================
     // Guardrails and Transformations
     //==========================================================================