Removed deprecated partition-file feature (#274)

* Removed deprecated partition-file feature
* Updated release notes
* Enforcing improved coverage after the recent test additions & removal of deprecated functions

Dockerfile

@@ -25,7 +25,6 @@ COPY ./src /assets/src
 COPY ./pom.xml /assets/pom.xml
 COPY ./src/resources/cdm.properties /assets/
 COPY ./src/resources/cdm-detailed.properties /assets/
-COPY ./src/resources/partitions.csv /assets/
 COPY ./scripts/get-latest-maven-version.sh ./get-latest-maven-version.sh
 
 RUN chmod +x ./get-latest-maven-version.sh && \

README.md

@@ -49,6 +49,13 @@ Exception in thread "main" java.lang.NoSuchMethodError: scala.runtime.Statics.re
 Note:
 - Above command generates a log file `logfile_name_*.txt` to avoid log output on the console.
 - Update the memory options (driver & executor memory) based on your use-case
+- To track details of a run in the `target` keyspace, pass param `--conf spark.cdm.trackRun=true`
+- To filter and migrate data only in a specific token range, you can pass the below two additional params to the `Migration` or `Validation` jobs 
+
+```
+--conf spark.cdm.filter.cassandra.partition.min=<token-range-min>
+--conf spark.cdm.filter.cassandra.partition.max=<token-range-max>
+```
 
 # Steps for Data-Validation:
 
@@ -84,7 +91,7 @@ Note:
 - The validation job will never delete records from target i.e. it only adds or updates data on target
 
 # Rerun (previously incomplete) Migration or Validation 
-- You can rerun a Migration or Validation job to complete a previous run that could have stopped for any reasons. This mode will skip any token-ranges from previous run that were migrated or validated successfully. This is done by passing the `spark.cdm.trackRun.previousRunId` param as shown below
+- You can rerun/resume a Migration or Validation job to complete a previous run that could have stopped (or completed with some errors) for any reasons. This mode will skip any token-ranges from the previous run that were migrated (or validated) successfully. This is done by passing the `spark.cdm.trackRun.previousRunId` param as shown below
 
 ```
 ./spark-submit --properties-file cdm.properties \
@@ -93,8 +100,6 @@ Note:
  --master "local[*]" --driver-memory 25G --executor-memory 25G \
  --class com.datastax.cdm.job.<Migrate|DiffData> cassandra-data-migrator-4.x.x.jar &> logfile_name_$(date +%Y%m%d_%H_%M).txt
 ```
-Note: 
-- This feature replaces and improves upon an older similar feature (using param `spark.cdm.tokenrange.partitionFile`) that is now deprecated and will be removed soon.
 
 # Perform large-field Guardrail violation checks
 - The tool can be used to identify large fields from a table that may break you cluster guardrails (e.g. AstraDB has a 10MB limit for a single large field)  `--class com.datastax.cdm.job.GuardrailCheck` as shown below
@@ -125,7 +130,6 @@ Note:
 - Validate migration accuracy and performance using a smaller randomized data-set
 - Supports adding custom fixed `writetime`
 - Track run information (start-time, end-time, status, etc.) in tables (`cdm_run_info` and `cdm_run_details`) on the target keyspace
-- Validation - Log partitions range level exceptions, use the exceptions file as input for rerun  
 
 # Things to know
 - Each run (Migration or Validation) can be tracked (when enabled). You can find summary and details of the same in tables `cdm_run_info` and `cdm_run_details` in the target keyspace.

RELEASE.md

@@ -1,4 +1,7 @@
 # Release Notes
+## [4.3.3] - 2024-07-22
+- Removed deprecated functionality related to processing token-ranges via partition-file 
+
 ## [4.3.2] - 2024-07-19
 - Removed deprecated functionality related to retry
 

pom.xml

@@ -277,7 +277,7 @@
 										<limit>
 											<counter>LINE</counter>
 											<value>MISSEDCOUNT</value>
-											<maximum>1500</maximum>
+											<maximum>1400</maximum>
 										</limit>
 									</limits>
 								</rule>

rat-excludes.txt

@@ -13,8 +13,6 @@ Dockerfile
 .github/**
 scripts/*
 test-backup/feature/*
-src/resources/partitions.csv
-src/resources/primary_key_rows.csv
 src/resources/log4j.xml
 SIT/*/*/*.assert
 SIT/*/*/*.out

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

@@ -33,83 +33,83 @@
 
 public abstract class AbstractJobSession<T> extends BaseJobSession {
 
-    public Logger logger = LoggerFactory.getLogger(this.getClass().getName());
-    protected EnhancedSession originSession;
-    protected EnhancedSession targetSession;
-    protected Guardrail guardrailFeature;
-    protected boolean guardrailEnabled;
-    protected boolean appendPartitionOnDiff = SplitPartitions.appendPartitionOnDiff(propertyHelper);
-    protected String partitionFileInput = SplitPartitions.getPartitionFileInput(propertyHelper);
-    protected String partitionFileOutput = SplitPartitions.getPartitionFileOutput(propertyHelper);
-    protected JobCounter jobCounter;
-    protected Long printStatsAfter;
-    protected Boolean trackRun = false;
-    protected TrackRun trackRunFeature;
-
-    protected AbstractJobSession(CqlSession originSession, CqlSession targetSession, SparkConf sc) {
-        this(originSession, targetSession, sc, false);
-    }
-
-    protected AbstractJobSession(CqlSession originSession, CqlSession targetSession, SparkConf sc, boolean isJobMigrateRowsFromFile) {
-        super(sc);
-
-        if (originSession == null) {
-            return;
-        }
-
-        this.printStatsAfter = propertyHelper.getLong(KnownProperties.PRINT_STATS_AFTER);
-        if (!propertyHelper.meetsMinimum(KnownProperties.PRINT_STATS_AFTER, printStatsAfter, 1L)) {
-            logger.warn(KnownProperties.PRINT_STATS_AFTER + " must be greater than 0.  Setting to default value of " + KnownProperties.getDefaultAsString(KnownProperties.PRINT_STATS_AFTER));
-            propertyHelper.setProperty(KnownProperties.PRINT_STATS_AFTER, KnownProperties.getDefault(KnownProperties.PRINT_STATS_AFTER));
-            printStatsAfter = propertyHelper.getLong(KnownProperties.PRINT_STATS_AFTER);
-        }
-        this.jobCounter = new JobCounter(printStatsAfter, propertyHelper.getBoolean(KnownProperties.PRINT_STATS_PER_PART));
-
-        rateLimiterOrigin = RateLimiter.create(propertyHelper.getInteger(KnownProperties.PERF_RATELIMIT_ORIGIN));
-        rateLimiterTarget = RateLimiter.create(propertyHelper.getInteger(KnownProperties.PERF_RATELIMIT_TARGET));
-        trackRun = propertyHelper.getBoolean(KnownProperties.TRACK_RUN);
-
-        logger.info("PARAM -- Partition file input: {}", partitionFileInput);
-        logger.info("PARAM -- Partition file output: {}", partitionFileOutput);
-        logger.info("PARAM -- Origin Rate Limit: {}", rateLimiterOrigin.getRate());
-        logger.info("PARAM -- Target Rate Limit: {}", rateLimiterTarget.getRate());
-
-        this.originSession = new EnhancedSession(propertyHelper, originSession, true);
-        this.targetSession = new EnhancedSession(propertyHelper, targetSession, false);
-        this.originSession.getCqlTable().setOtherCqlTable(this.targetSession.getCqlTable());
-        this.targetSession.getCqlTable().setOtherCqlTable(this.originSession.getCqlTable());
-        this.originSession.getCqlTable().setFeatureMap(featureMap);
-        this.targetSession.getCqlTable().setFeatureMap(featureMap);
-
-        boolean allFeaturesValid = true;
-        for (Feature f : featureMap.values()) {
-            if (!f.initializeAndValidate(this.originSession.getCqlTable(), this.targetSession.getCqlTable())) {
-                allFeaturesValid = false;
-                logger.error("Feature {} is not valid.  Please check the configuration.", f.getClass().getName());
-            }
-        }
-        if (!allFeaturesValid) {
-            throw new RuntimeException("One or more features are not valid.  Please check the configuration.");
-        }
-
-        PKFactory pkFactory = new PKFactory(propertyHelper, this.originSession.getCqlTable(), this.targetSession.getCqlTable());
-        this.originSession.setPKFactory(pkFactory);
-        this.targetSession.setPKFactory(pkFactory);
-
-        // Guardrail is referenced by many jobs, and is evaluated against the target table
-        this.guardrailFeature = (Guardrail) this.targetSession.getCqlTable().getFeature(Featureset.GUARDRAIL_CHECK);
-        this.guardrailEnabled = this.guardrailFeature.isEnabled();
-    }
-
-    public abstract void processSlice(T slice);
-    
-    public synchronized void initCdmRun(Collection<SplitPartitions.Partition> parts, TrackRun trackRunFeature) {}
-
-    public synchronized void printCounts(boolean isFinal) {
-        if (isFinal) {
-            jobCounter.printFinal(trackRun, trackRunFeature);
-        } else {
-            jobCounter.printProgress();
-        }
-    }
+	public Logger logger = LoggerFactory.getLogger(this.getClass().getName());
+	protected EnhancedSession originSession;
+	protected EnhancedSession targetSession;
+	protected Guardrail guardrailFeature;
+	protected boolean guardrailEnabled;
+	protected JobCounter jobCounter;
+	protected Long printStatsAfter;
+	protected TrackRun trackRunFeature;
+
+	protected AbstractJobSession(CqlSession originSession, CqlSession targetSession, SparkConf sc) {
+		this(originSession, targetSession, sc, false);
+	}
+
+	protected AbstractJobSession(CqlSession originSession, CqlSession targetSession, SparkConf sc,
+			boolean isJobMigrateRowsFromFile) {
+		super(sc);
+
+		if (originSession == null) {
+			return;
+		}
+
+		this.printStatsAfter = propertyHelper.getLong(KnownProperties.PRINT_STATS_AFTER);
+		if (!propertyHelper.meetsMinimum(KnownProperties.PRINT_STATS_AFTER, printStatsAfter, 1L)) {
+			logger.warn(KnownProperties.PRINT_STATS_AFTER + " must be greater than 0.  Setting to default value of "
+					+ KnownProperties.getDefaultAsString(KnownProperties.PRINT_STATS_AFTER));
+			propertyHelper.setProperty(KnownProperties.PRINT_STATS_AFTER,
+					KnownProperties.getDefault(KnownProperties.PRINT_STATS_AFTER));
+			printStatsAfter = propertyHelper.getLong(KnownProperties.PRINT_STATS_AFTER);
+		}
+		this.jobCounter = new JobCounter(printStatsAfter,
+				propertyHelper.getBoolean(KnownProperties.PRINT_STATS_PER_PART));
+
+		rateLimiterOrigin = RateLimiter.create(propertyHelper.getInteger(KnownProperties.PERF_RATELIMIT_ORIGIN));
+		rateLimiterTarget = RateLimiter.create(propertyHelper.getInteger(KnownProperties.PERF_RATELIMIT_TARGET));
+
+		logger.info("PARAM -- Origin Rate Limit: {}", rateLimiterOrigin.getRate());
+		logger.info("PARAM -- Target Rate Limit: {}", rateLimiterTarget.getRate());
+
+		this.originSession = new EnhancedSession(propertyHelper, originSession, true);
+		this.targetSession = new EnhancedSession(propertyHelper, targetSession, false);
+		this.originSession.getCqlTable().setOtherCqlTable(this.targetSession.getCqlTable());
+		this.targetSession.getCqlTable().setOtherCqlTable(this.originSession.getCqlTable());
+		this.originSession.getCqlTable().setFeatureMap(featureMap);
+		this.targetSession.getCqlTable().setFeatureMap(featureMap);
+
+		boolean allFeaturesValid = true;
+		for (Feature f : featureMap.values()) {
+			if (!f.initializeAndValidate(this.originSession.getCqlTable(), this.targetSession.getCqlTable())) {
+				allFeaturesValid = false;
+				logger.error("Feature {} is not valid.  Please check the configuration.", f.getClass().getName());
+			}
+		}
+		if (!allFeaturesValid) {
+			throw new RuntimeException("One or more features are not valid.  Please check the configuration.");
+		}
+
+		PKFactory pkFactory = new PKFactory(propertyHelper, this.originSession.getCqlTable(),
+				this.targetSession.getCqlTable());
+		this.originSession.setPKFactory(pkFactory);
+		this.targetSession.setPKFactory(pkFactory);
+
+		// Guardrail is referenced by many jobs, and is evaluated against the target
+		// table
+		this.guardrailFeature = (Guardrail) this.targetSession.getCqlTable().getFeature(Featureset.GUARDRAIL_CHECK);
+		this.guardrailEnabled = this.guardrailFeature.isEnabled();
+	}
+
+	public abstract void processSlice(T slice);
+
+	public synchronized void initCdmRun(Collection<SplitPartitions.Partition> parts, TrackRun trackRunFeature) {
+	}
+
+	public synchronized void printCounts(boolean isFinal) {
+		if (isFinal) {
+			jobCounter.printFinal(trackRunFeature);
+		} else {
+			jobCounter.printProgress();
+		}
+	}
 }