Merge pull request #42 from datastax/feature/config-should-represent-base-use-case

pravinbhat · web-flow · commit 69bb9ecc50dc · 2022-12-11T23:30:57.000-05:00
Refactorred config
diff --git a/.github/workflows/maven.yml b/.github/workflows/maven.yml
@@ -0,0 +1,32 @@
+# This workflow will build a Java project with Maven, and cache/restore any dependencies to improve the workflow execution time
+# For more information see: https://docs.github.com/en/actions/automating-builds-and-tests/building-and-testing-java-with-maven
+
+# This workflow uses actions that are not certified by GitHub.
+# They are provided by a third-party and are governed by
+# separate terms of service, privacy policy, and support
+# documentation.
+
+name: Java CI with Maven
+
+on: [push, pull_request]
+
+jobs:
+  build:
+
+    runs-on: ubuntu-latest
+
+    steps:
+    - uses: actions/checkout@v3
+    - name: Set up JDK 8
+      uses: actions/setup-java@v3
+      with:
+        java-version: '8'
+        distribution: 'temurin'
+        cache: maven
+    - name: Build with Maven
+      run: mvn -B package --file pom.xml
+
+    # Optional: Uploads the full dependency graph to GitHub to improve the quality of Dependabot alerts this repository can receive
+    - name: Update dependency graph
+      uses: advanced-security/maven-dependency-submission-action@571e99aab1055c2e71a1e2309b9691de18d6b7d6
+
diff --git a/README.md b/README.md
@@ -1,6 +1,6 @@
 # cassandra-data-migrator
 
-Spark jobs in this repo can be used for data migration and data validation.
+Migrate and Validate Tables between Origin and Target Cassandra Clusters.
 
 > :warning: Please note this job has been tested with spark version [2.4.8](https://archive.apache.org/dist/spark/spark-2.4.8/)
 
@@ -20,12 +20,12 @@ wget https://downloads.apache.org/spark/spark-2.4.8/
 tar -xvzf <spark downloaded file name>
 ```
 
-# Steps:
+# Steps for Data-Migration:
 
 1. `sparkConf.properties` file needs to be configured as applicable for the environment
    > A sample Spark conf file configuration can be [found here](./src/resources/sparkConf.properties)
 2. Place the conf file where it can be accessed while running the job via spark-submit.
-3. Run the 'Data Migration' job using `spark-submit` command as shown below:
+3. Run the below job using `spark-submit` command as shown below:
 
 ```
 ./spark-submit --properties-file sparkConf.properties /
@@ -36,7 +36,7 @@ tar -xvzf <spark downloaded file name>
 Note: Above command also generates a log file `logfile_name.txt` to avoid log output on the console.
 
 
-# Data-validation job
+# Steps for Data-Validation:
 
 - To run the job in Data validation mode, use class option `--class datastax.astra.migrate.DiffData` as shown below
 
@@ -90,6 +90,6 @@ This mode is specifically useful to processes a subset of partition-ranges that
 - Advanced DataTypes ([Sets](https://docs.datastax.com/en/dse/6.8/cql/cql/cql_reference/refDataTypes.html#refDataTypes__set), [Lists](https://docs.datastax.com/en/dse/6.8/cql/cql/cql_reference/refDataTypes.html#refDataTypes__list), [Maps](https://docs.datastax.com/en/dse/6.8/cql/cql/cql_reference/refDataTypes.html#refDataTypes__map), [UDTs](https://docs.datastax.com/en/dse/6.8/cql/cql/cql_reference/refDataTypes.html#refDataTypes__udt))
 - Filter records from origin using writetime
 - SSL Support (including custom cipher algorithms)
-- Migrate from any Cassandra origin ([Apache Cassandra](https://cassandra.apache.org)/[DataStax Enterprise (DSE)](https://www.datastax.com/products/datastax-enterprise)/[DataStax Astra DB](https://www.datastax.com/products/datastax-astra)) to any Cassandra target ([Apache Cassandra](https://cassandra.apache.org)/[DataStax Enterprise (DSE)](https://www.datastax.com/products/datastax-enterprise)/[DataStax Astra DB](https://www.datastax.com/products/datastax-astra))
+- Migrate from any Cassandra origin ([Apache Cassandra](https://cassandra.apache.org) / [DataStax Enterprise](https://www.datastax.com/products/datastax-enterprise) / [DataStax Astra DB](https://www.datastax.com/products/datastax-astra)) to any Cassandra target ([Apache Cassandra](https://cassandra.apache.org) / [DataStax Enterprise](https://www.datastax.com/products/datastax-enterprise) / [DataStax Astra DB](https://www.datastax.com/products/datastax-astra))
 - Validate migration accuracy and performance using a smaller randomized data-set
 - Custom writetime
diff --git a/cassandra-data-migrator.iml b/cassandra-data-migrator.iml
@@ -197,6 +197,17 @@
     <orderEntry type="library" name="Maven: com.github.spotbugs:spotbugs-annotations:3.1.12" level="project" />
     <orderEntry type="library" name="Maven: com.datastax.oss:java-driver-mapper-runtime:4.10.0" level="project" />
     <orderEntry type="library" name="Maven: com.datastax.oss:java-driver-query-builder:4.10.0" level="project" />
+    <orderEntry type="library" name="Maven: com.github.jnr:jnr-posix:3.1.15" level="project" />
+    <orderEntry type="library" name="Maven: com.github.jnr:jnr-ffi:2.2.11" level="project" />
+    <orderEntry type="library" name="Maven: com.github.jnr:jffi:1.3.9" level="project" />
+    <orderEntry type="library" scope="RUNTIME" name="Maven: com.github.jnr:jffi:native:1.3.9" level="project" />
+    <orderEntry type="library" name="Maven: org.ow2.asm:asm-commons:9.2" level="project" />
+    <orderEntry type="library" name="Maven: org.ow2.asm:asm-analysis:9.2" level="project" />
+    <orderEntry type="library" name="Maven: org.ow2.asm:asm-tree:9.2" level="project" />
+    <orderEntry type="library" name="Maven: org.ow2.asm:asm-util:9.2" level="project" />
+    <orderEntry type="library" name="Maven: com.github.jnr:jnr-a64asm:1.0.0" level="project" />
+    <orderEntry type="library" name="Maven: com.github.jnr:jnr-x86asm:1.0.2" level="project" />
+    <orderEntry type="library" name="Maven: com.github.jnr:jnr-constants:0.10.3" level="project" />
     <orderEntry type="library" name="Maven: org.apache.logging.log4j:log4j-api:2.19.0" level="project" />
     <orderEntry type="library" name="Maven: org.apache.logging.log4j:log4j-core:2.19.0" level="project" />
     <orderEntry type="library" name="Maven: org.apache.logging.log4j:log4j-to-slf4j:2.19.0" level="project" />
@@ -248,7 +259,7 @@
     <orderEntry type="library" scope="TEST" name="Maven: org.caffinitas.ohc:ohc-core-j8:0.4.4" level="project" />
     <orderEntry type="library" scope="TEST" name="Maven: com.github.ben-manes.caffeine:caffeine:2.2.6" level="project" />
     <orderEntry type="library" scope="TEST" name="Maven: org.jctools:jctools-core:1.2.1" level="project" />
-    <orderEntry type="library" scope="TEST" name="Maven: org.ow2.asm:asm:5.0.4" level="project" />
+    <orderEntry type="library" name="Maven: org.ow2.asm:asm:5.0.4" level="project" />
     <orderEntry type="library" scope="TEST" name="Maven: javax.inject:javax.inject:1" level="project" />
     <orderEntry type="library" name="Maven: org.hdrhistogram:HdrHistogram:2.1.9" level="project" />
     <orderEntry type="library" scope="TEST" name="Maven: de.jflex:jflex:1.6.0" level="project" />
diff --git a/pom.xml b/pom.xml
@@ -3,7 +3,7 @@
 
   <groupId>datastax.astra.migrate</groupId>
   <artifactId>cassandra-data-migrator</artifactId>
-  <version>2.8</version>
+  <version>2.9</version>
   <packaging>jar</packaging>
 
   <properties>
@@ -72,6 +72,11 @@
       <artifactId>spark-cassandra-connector_${scala.main.version}</artifactId>
       <version>${connector.version}</version>
     </dependency>
+    <dependency>
+      <groupId>com.github.jnr</groupId>
+      <artifactId>jnr-posix</artifactId>
+      <version>3.1.15</version>
+    </dependency>
 
     <dependency>
       <groupId>org.apache.logging.log4j</groupId>
diff --git a/src/main/java/datastax/astra/migrate/AbstractJobSession.java b/src/main/java/datastax/astra/migrate/AbstractJobSession.java
@@ -11,6 +11,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.time.Duration;
+import java.time.Instant;
 import java.util.Map;
 import java.util.stream.IntStream;
 
@@ -84,9 +85,15 @@ protected AbstractJobSession(CqlSession sourceSession, CqlSession astraSession,
         logger.info("PARAM -- Destination Keyspace Table: {}", astraKeyspaceTable);
         logger.info("PARAM -- ReadRateLimit: {}", readLimiter.getRate());
         logger.info("PARAM -- WriteRateLimit: {}", writeLimiter.getRate());
-        logger.info("PARAM -- TTLCols: {}" + ttlCols);
+        logger.info("PARAM -- TTLCols: {}", ttlCols);
         logger.info("PARAM -- WriteTimestampFilterCols: {}", writeTimeStampCols);
         logger.info("PARAM -- WriteTimestampFilter: {}", writeTimeStampFilter);
+        if (writeTimeStampFilter) {
+            logger.info("PARAM -- minWriteTimeStampFilter: {} datetime is {}", minWriteTimeStampFilter,
+                    Instant.ofEpochMilli(minWriteTimeStampFilter / 1000));
+            logger.info("PARAM -- maxWriteTimeStampFilter: {} datetime is {}", maxWriteTimeStampFilter,
+                    Instant.ofEpochMilli(maxWriteTimeStampFilter / 1000));
+        }
 
         String selectCols = Util.getSparkProp(sc, "spark.query.origin");
         String partionKey = Util.getSparkProp(sc, "spark.query.origin.partitionKey");
diff --git a/src/main/java/datastax/astra/migrate/CopyJobSession.java b/src/main/java/datastax/astra/migrate/CopyJobSession.java
@@ -61,11 +61,12 @@ public void getDataAndInsert(BigInteger min, BigInteger max) {
                         if (filterData) {
                             String col = (String) getData(new MigrateDataType(filterColType), filterColIndex, sourceRow);
                             if (col.trim().equalsIgnoreCase(filterColValue)) {
-                                logger.warn("Skipping row and filtering out: " + getKey(sourceRow));
+                                logger.warn("Skipping row and filtering out: {}", getKey(sourceRow));
                                 skippedCounter.incrementAndGet();
                                 continue;
                             }
                         }
+
                         if (writeTimeStampFilter) {
                             // only process rows greater than writeTimeStampFilter
                             Long sourceWriteTimeStamp = getLargestWriteTimeStamp(sourceRow);
@@ -111,7 +112,7 @@ public void getDataAndInsert(BigInteger min, BigInteger max) {
                         if (filterData) {
                             String colValue = (String) getData(new MigrateDataType(filterColType), filterColIndex, sourceRow);
                             if (colValue.trim().equalsIgnoreCase(filterColValue)) {
-                                logger.warn("Skipping row and filtering out: " + getKey(sourceRow));
+                                logger.warn("Skipping row and filtering out: {}", getKey(sourceRow));
                                 skippedCounter.incrementAndGet();
                                 continue;
                             }
diff --git a/src/main/scala/datastax/astra/migrate/AbstractJob.scala b/src/main/scala/datastax/astra/migrate/AbstractJob.scala
@@ -10,13 +10,13 @@ class AbstractJob extends BaseJob {
   abstractLogger.info("PARAM -- Split Size: " + splitSize)
   abstractLogger.info("PARAM -- Coverage Percent: " + coveragePercent)
 
-  var sourceConnection = getConnection(true, sourceIsAstra, sourceScbPath, sourceHost, sourceUsername, sourcePassword,
+  var sourceConnection = getConnection(true, sourceScbPath, sourceHost, sourceUsername, sourcePassword,
     sourceTrustStorePath, sourceTrustStorePassword, sourceTrustStoreType, sourceKeyStorePath, sourceKeyStorePassword, sourceEnabledAlgorithms);
 
-  var destinationConnection = getConnection(false, destinationIsAstra, destinationScbPath, destinationHost, destinationUsername, destinationPassword,
+  var destinationConnection = getConnection(false, destinationScbPath, destinationHost, destinationUsername, destinationPassword,
     destinationTrustStorePath, destinationTrustStorePassword, destinationTrustStoreType, destinationKeyStorePath, destinationKeyStorePassword, destinationEnabledAlgorithms);
 
-  private def getConnection(isSource: Boolean, isAstra: String, scbPath: String, host: String, username: String, password: String,
+  private def getConnection(isSource: Boolean, scbPath: String, host: String, username: String, password: String,
                             trustStorePath: String, trustStorePassword: String, trustStoreType: String,
                             keyStorePath: String, keyStorePassword: String, enabledAlgorithms: String): CassandraConnector = {
     var connType: String = "Source"
@@ -25,15 +25,15 @@ class AbstractJob extends BaseJob {
     }
 
     var config: SparkConf = sContext.getConf
-    if ("true".equals(isAstra)) {
+    if (scbPath.nonEmpty) {
       abstractLogger.info(connType + ": Connecting to Astra using SCB: " + scbPath);
 
       return CassandraConnector(config
         .set("spark.cassandra.auth.username", username)
         .set("spark.cassandra.auth.password", password)
         .set("spark.cassandra.input.consistency.level", consistencyLevel)
         .set("spark.cassandra.connection.config.cloud.path", scbPath))
-    } else if (null != trustStorePath && !trustStorePath.trim.isEmpty) {
+    } else if (trustStorePath.nonEmpty) {
       abstractLogger.info(connType + ": Connecting to Cassandra (or DSE) with SSL host: " + host);
 
       // Use defaults when not provided
diff --git a/src/main/scala/datastax/astra/migrate/BaseJob.scala b/src/main/scala/datastax/astra/migrate/BaseJob.scala
@@ -20,7 +20,6 @@ class BaseJob extends App {
 
   val consistencyLevel = Util.getSparkPropOr(sc, "spark.read.consistency.level", "LOCAL_QUORUM")
 
-  val sourceIsAstra = Util.getSparkPropOr(sc, "spark.origin.isAstra", "false")
   val sourceScbPath = Util.getSparkPropOrEmpty(sc, "spark.origin.scb")
   val sourceHost = Util.getSparkPropOrEmpty(sc, "spark.origin.host")
   val sourceUsername = Util.getSparkPropOrEmpty(sc, "spark.origin.username")
@@ -32,7 +31,6 @@ class BaseJob extends App {
   val sourceKeyStorePassword = Util.getSparkPropOrEmpty(sc, "spark.origin.keyStore.password")
   val sourceEnabledAlgorithms = Util.getSparkPropOrEmpty(sc, "spark.origin.enabledAlgorithms")
 
-  val destinationIsAstra = Util.getSparkPropOr(sc, "spark.target.isAstra", "true")
   val destinationScbPath = Util.getSparkPropOrEmpty(sc, "spark.target.scb")
   val destinationHost = Util.getSparkPropOrEmpty(sc, "spark.target.host")
   val destinationUsername = Util.getSparkProp(sc, "spark.target.username")
diff --git a/src/main/scala/datastax/astra/migrate/OriginData.scala b/src/main/scala/datastax/astra/migrate/OriginData.scala
@@ -9,26 +9,26 @@ object OriginData extends BaseJob {
 
   val logger = LoggerFactory.getLogger(this.getClass.getName)
   logger.info("Started Migration App")
-  var sourceConnection = getConnection(true, sourceIsAstra, sourceScbPath, sourceHost, sourceUsername, sourcePassword,
+  var sourceConnection = getConnection(true, sourceScbPath, sourceHost, sourceUsername, sourcePassword,
     sourceTrustStorePath, sourceTrustStorePassword, sourceTrustStoreType, sourceKeyStorePath, sourceKeyStorePassword, sourceEnabledAlgorithms);
   analyzeSourceTable(sourceConnection)
   exitSpark
 
 
-  private def getConnection(isSource: Boolean, isAstra: String, scbPath: String, host: String, username: String, password: String,
+  private def getConnection(isSource: Boolean, scbPath: String, host: String, username: String, password: String,
                             trustStorePath: String, trustStorePassword: String, trustStoreType: String,
                             keyStorePath: String, keyStorePassword: String, enabledAlgorithms: String): CassandraConnector = {
     var connType: String = "Source"
 
-    if ("true".equals(isAstra)) {
+    if (scbPath.nonEmpty) {
       abstractLogger.info(connType + ": Connected to Astra!");
 
       return CassandraConnector(sc
         .set("spark.cassandra.auth.username", username)
         .set("spark.cassandra.auth.password", password)
         .set("spark.cassandra.input.consistency.level", consistencyLevel)
         .set("spark.cassandra.connection.config.cloud.path", scbPath))
-    } else if (null != trustStorePath && !trustStorePath.trim.isEmpty) {
+    } else if (trustStorePath.nonEmpty) {
       abstractLogger.info(connType + ": Connected to Cassandra (or DSE) with SSL!");
 
       // Use defaults when not provided
diff --git a/src/resources/sparkConf.properties b/src/resources/sparkConf.properties
