Support Azure/cosmos compatibility (needed SSL enable support)

Author: pravinbhat

pom.xml

@@ -3,7 +3,7 @@
 
   <groupId>datastax.astra.migrate</groupId>
   <artifactId>cassandra-data-migrator</artifactId>
-  <version>3.0.5</version>
+  <version>3.0.6</version>
   <packaging>jar</packaging>
 
   <properties>

src/main/scala/datastax/astra/migrate/AbstractJob.scala

@@ -9,15 +9,17 @@ class AbstractJob extends BaseJob {
   abstractLogger.info("PARAM -- Max Partition: " + maxPartition)
   abstractLogger.info("PARAM -- Split Size: " + splitSize)
   abstractLogger.info("PARAM -- Coverage Percent: " + coveragePercent)
+  abstractLogger.info("PARAM -- Origin SSL Enabled: {}", sourceSSLEnabled);
+  abstractLogger.info("PARAM -- Target SSL Enabled: {}", destinationSSLEnabled);
 
-  var sourceConnection = getConnection(true, sourceScbPath, sourceHost, sourceUsername, sourcePassword,
+  var sourceConnection = getConnection(true, sourceScbPath, sourceHost, sourceUsername, sourcePassword, sourceSSLEnabled,
     sourceTrustStorePath, sourceTrustStorePassword, sourceTrustStoreType, sourceKeyStorePath, sourceKeyStorePassword, sourceEnabledAlgorithms);
 
-  var destinationConnection = getConnection(false, destinationScbPath, destinationHost, destinationUsername, destinationPassword,
+  var destinationConnection = getConnection(false, destinationScbPath, destinationHost, destinationUsername, destinationPassword, destinationSSLEnabled,
     destinationTrustStorePath, destinationTrustStorePassword, destinationTrustStoreType, destinationKeyStorePath, destinationKeyStorePassword, destinationEnabledAlgorithms);
 
   private def getConnection(isSource: Boolean, scbPath: String, host: String, username: String, password: String,
-                            trustStorePath: String, trustStorePassword: String, trustStoreType: String,
+                            sslEnabled: String, trustStorePath: String, trustStorePassword: String, trustStoreType: String,
                             keyStorePath: String, keyStorePassword: String, enabledAlgorithms: String): CassandraConnector = {
     var connType: String = "Source"
     if (!isSource) {
@@ -34,7 +36,7 @@ class AbstractJob extends BaseJob {
         .set("spark.cassandra.input.consistency.level", consistencyLevel)
         .set("spark.cassandra.connection.config.cloud.path", scbPath))
     } else if (trustStorePath.nonEmpty) {
-      abstractLogger.info(connType + ": Connecting to Cassandra (or DSE) with SSL host: " + host);
+      abstractLogger.info(connType + ": Connecting (with clientAuth) to Cassandra (or DSE) host: " + host);
 
       // Use defaults when not provided
       var enabledAlgorithmsVar = enabledAlgorithms
@@ -60,6 +62,7 @@ class AbstractJob extends BaseJob {
       abstractLogger.info(connType + ": Connecting to Cassandra (or DSE) host: " + host);
 
       return CassandraConnector(config.set("spark.cassandra.auth.username", username)
+        .set("spark.cassandra.connection.ssl.enabled", sslEnabled)
         .set("spark.cassandra.auth.password", password)
         .set("spark.cassandra.input.consistency.level", consistencyLevel)
         .set("spark.cassandra.connection.host", host))

src/main/scala/datastax/astra/migrate/BaseJob.scala

@@ -24,6 +24,7 @@ class BaseJob extends App {
   val sourceHost = Util.getSparkPropOrEmpty(sc, "spark.origin.host")
   val sourceUsername = Util.getSparkPropOrEmpty(sc, "spark.origin.username")
   val sourcePassword = Util.getSparkPropOrEmpty(sc, "spark.origin.password")
+  val sourceSSLEnabled = Util.getSparkPropOr(sc, "spark.origin.ssl.enabled", "false")
   val sourceTrustStorePath = Util.getSparkPropOrEmpty(sc, "spark.origin.trustStore.path")
   val sourceTrustStorePassword = Util.getSparkPropOrEmpty(sc, "spark.origin.trustStore.password")
   val sourceTrustStoreType = Util.getSparkPropOr(sc, "spark.origin.trustStore.type", "JKS")
@@ -35,6 +36,7 @@ class BaseJob extends App {
   val destinationHost = Util.getSparkPropOrEmpty(sc, "spark.target.host")
   val destinationUsername = Util.getSparkProp(sc, "spark.target.username")
   val destinationPassword = Util.getSparkProp(sc, "spark.target.password")
+  val destinationSSLEnabled = Util.getSparkPropOr(sc, "spark.target.ssl.enabled", "false")
   val destinationTrustStorePath = Util.getSparkPropOrEmpty(sc, "spark.target.trustStore.path")
   val destinationTrustStorePassword = Util.getSparkPropOrEmpty(sc, "spark.target.trustStore.password")
   val destinationTrustStoreType = Util.getSparkPropOr(sc, "spark.target.trustStore.type", "JKS")

src/resources/sparkConf.properties

@@ -114,6 +114,9 @@ spark.query.writetime.cols                        2,3
 #spark.origin.FilterColumnType                     6%16
 #spark.origin.FilterColumnValue                    test
 
+# ONLY USE if SSL is enabled on origin Cassandra/DSE
+#spark.origin.ssl.enabled                          true
+
 # ONLY USE if SSL clientAuth is enabled on origin Cassandra/DSE
 #spark.origin.trustStore.path
 #spark.origin.trustStore.password
@@ -122,6 +125,9 @@ spark.query.writetime.cols                        2,3
 #spark.origin.keyStore.password
 #spark.origin.enabledAlgorithms                   TLS_RSA_WITH_AES_128_CBC_SHA,TLS_RSA_WITH_AES_256_CBC_SHA
 
+# ONLY USE if SSL is enabled on target Cassandra/DSE
+#spark.target.ssl.enabled                          true
+
 # ONLY USE if SSL clientAuth is enabled on target Cassandra/DSE
 #spark.target.trustStore.path
 #spark.target.trustStore.password