Bump Spark Cassandra Connector to 3.4.1

Previously, Spark Cassandra Connector and Spark Core was using metrics-core 3.x and 4.x respectively. Bumping Cassandra Connector integrates a newer Cassandra Java Driver, so it does not interfere with spark's metrics-core version. Fixes #138 Signed-off-by: Vladimir Sitnikov <[email protected]>
jaegertracing · Feb 5, 2025 · 02f6bb1 · 02f6bb1
1 parent 2c61c38
commit 02f6bb1
Show file tree

Hide file tree

Showing 5 changed files with 45 additions and 68 deletions.
diff --git a/jaeger-spark-dependencies-cassandra/pom.xml b/jaeger-spark-dependencies-cassandra/pom.xml
@@ -25,7 +25,7 @@
   <artifactId>jaeger-spark-dependencies-cassandra</artifactId>
 
   <properties>
-    <spark-cassandra-connector.version>2.4.3</spark-cassandra-connector.version>
+    <spark-cassandra-connector.version>3.4.1</spark-cassandra-connector.version>
   </properties>
 
   <dependencies>
@@ -47,7 +47,7 @@
 
     <dependency>
       <groupId>com.datastax.spark</groupId>
-      <artifactId>spark-cassandra-connector-unshaded_${version.scala.binary}</artifactId>
+      <artifactId>spark-cassandra-connector_${version.scala.binary}</artifactId>
       <version>${spark-cassandra-connector.version}</version>
     </dependency>
 
@@ -72,6 +72,15 @@
       <artifactId>testcontainers</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.testcontainers</groupId>
+      <artifactId>cassandra</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>com.google.guava</groupId>
+      <artifactId>guava</artifactId>
+    </dependency>
     <dependency>
       <groupId>com.codahale.metrics</groupId>
       <artifactId>metrics-core</artifactId>

diff --git a/...src/main/java/io/jaegertracing/spark/dependencies/cassandra/CassandraDependenciesJob.java b/...src/main/java/io/jaegertracing/spark/dependencies/cassandra/CassandraDependenciesJob.java
@@ -66,6 +66,8 @@ public static final class Builder {
     String localDc = Utils.getEnv("CASSANDRA_LOCAL_DC", null);
     // local[*] master lets us run & test the job locally without setting a Spark cluster
     String sparkMaster = Utils.getEnv("SPARK_MASTER", "local[*]");
+    String username = Utils.getEnv("CASSANDRA_USERNAME", "");
+    String password = Utils.getEnv("CASSANDRA_PASSWORD", "");
     // needed when not in local mode
     String[] jars;
 
@@ -88,8 +90,6 @@ public static final class Builder {
     		  System.getProperty("javax.net.ssl.keyStore", ""));
       sparkProperties.put("spark.cassandra.connection.ssl.keyStore.password", 
     		  System.getProperty("javax.net.ssl.keyStorePassword", ""));
-      sparkProperties.put("spark.cassandra.auth.username", Utils.getEnv("CASSANDRA_USERNAME", ""));
-      sparkProperties.put("spark.cassandra.auth.password", Utils.getEnv("CASSANDRA_PASSWORD", ""));
     }
 
     /** When set, this indicates which jars to distribute to the cluster. */
@@ -105,6 +105,20 @@ public Builder keyspace(String keyspace) {
       return this;
     }
 
+    /** Cassandra username. */
+    public Builder username(String username) {
+      Utils.checkNoTNull("username", username);
+      this.username = username;
+      return this;
+    }
+
+    /** Cassandra username. */
+    public Builder password(String password) {
+      Utils.checkNoTNull("password", password);
+      this.password = password;
+      return this;
+    }
+
     /** Day to process dependencies for. Defaults to today. */
     public Builder day(LocalDate day) {
       this.day = day.atStartOfDay(ZoneOffset.UTC);
@@ -143,6 +157,8 @@ public CassandraDependenciesJob build() {
         .setAppName(getClass().getName());
     conf.set("spark.cassandra.connection.host", parseHosts(builder.contactPoints));
     conf.set("spark.cassandra.connection.port", parsePort(builder.contactPoints));
+    conf.set("spark.cassandra.auth.username", builder.username);
+    conf.set("spark.cassandra.auth.password", builder.password);
     if (builder.localDc != null) {
       conf.set("connection.local_dc", builder.localDc);
     }
@@ -196,7 +212,7 @@ static String parseHosts(String contactPoints) {
     List<String> result = new LinkedList<>();
     for (String contactPoint : contactPoints.split(",")) {
       HostAndPort parsed = HostAndPort.fromString(contactPoint);
-      result.add(parsed.getHostText());
+      result.add(parsed.getHost());
     }
     return Joiner.on(',').join(result);
   }

diff --git a/...andra/src/test/java/io/jaegertracing/spark/dependencies/cassandra/CassandraContainer.java b/...andra/src/test/java/io/jaegertracing/spark/dependencies/cassandra/CassandraContainer.java
diff --git a/...test/java/io/jaegertracing/spark/dependencies/cassandra/CassandraDependenciesJobTest.java b/...test/java/io/jaegertracing/spark/dependencies/cassandra/CassandraDependenciesJobTest.java
@@ -26,6 +26,7 @@
 import org.junit.After;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
+import org.testcontainers.containers.CassandraContainer;
 import org.testcontainers.containers.GenericContainer;
 import org.testcontainers.containers.Network;
 import org.testcontainers.containers.wait.strategy.HttpWaitStrategy;
@@ -45,7 +46,7 @@ public class CassandraDependenciesJobTest extends DependenciesTest {
   @BeforeClass
   public static void beforeClass() {
     network = Network.newNetwork();
-    cassandra = new CassandraContainer("cassandra:4.1")
+    cassandra = new CassandraContainer<>("cassandra:4.1")
         .withNetwork(network)
         .withNetworkAliases("cassandra")
         .withExposedPorts(9042);
@@ -117,6 +118,8 @@ protected void deriveDependencies() {
         .contactPoints("localhost:" + cassandraPort)
         .day(LocalDate.now())
         .keyspace("jaeger_v1_dc1")
+        .username(cassandra.getUsername())
+        .password(cassandra.getPassword())
         .build()
         .run("peer.service");
   }

diff --git a/pom.xml b/pom.xml
@@ -44,7 +44,7 @@
     <version.io.zipkin.reporter-zipkin-sender-okhttp3>1.0.2</version.io.zipkin.reporter-zipkin-sender-okhttp3>
     <version.junit>4.13.2</version.junit>
     <version.org.assertj>3.24.2</version.org.assertj>
-    <version.org.testcontainers>1.18.1</version.org.testcontainers>
+    <version.org.testcontainers>1.20.4</version.org.testcontainers>
     <version.com.squareup.okhttp3-okhttp>3.14.9</version.com.squareup.okhttp3-okhttp>
     <version.org.awaitility-awaitility>4.2.0</version.org.awaitility-awaitility>
 
@@ -137,6 +137,16 @@
         <artifactId>testcontainers</artifactId>
         <version>${version.org.testcontainers}</version>
       </dependency>
+      <dependency>
+        <groupId>org.testcontainers</groupId>
+        <artifactId>cassandra</artifactId>
+        <version>${version.org.testcontainers}</version>
+      </dependency>
+      <dependency>
+        <groupId>com.google.guava</groupId>
+        <artifactId>guava</artifactId>
+        <version>27.0-jre</version>
+      </dependency>
       <!-- Forcibly bump Commons Collection version to avoid CVE-2015-7501 -->
       <dependency>
         <groupId>commons-collections</groupId>