CASSSIDECAR-308 CDC: Add end-to-end CDC integration tests

CHANGES.txt

@@ -1,5 +1,6 @@
 0.3.0
 -----
+ * CDC: Add end-to-end CDC integration tests (CASSSIDECAR-308)
  * SchemaStorePublisherFactory should be Injectable in CachingSchemaStore (CASSSIDECAR-408)
  * Fix StorageClientTest Docker API compatibility and improve CI test reporting (CASSSIDECAR-410)
  * Incorrect SSL Configuration Keys in CdcPublisher.secretsProvider() (CASSSIDECAR-401)

gradle.properties

@@ -46,5 +46,5 @@ swaggerVersion=2.2.21
 kryoVersion=4.0.2
 # OSHI dependencies
 oshiVersion=6.9.0
-analyticsVersion=0.2.0
+analyticsVersion=0.3.0
 kafkaClientVersion=3.7.0

integration-framework/build.gradle

@@ -65,9 +65,16 @@ dependencies {
     api("io.vertx:vertx-junit5:${project.vertxVersion}")
     // The server itself
     api(project(path: ":server"))
+    api(testFixtures(project(path: ":server")))
     api(project(path: ":server-common"))
+
+    // CDC dependencies
+    api(group: "org.apache.cassandra", name: "cassandra-analytics-cdc_spark3_2.12", version: "${project.analyticsVersion}")
+    api(group: "org.apache.cassandra", name: "cassandra-analytics-cdc-sidecar_spark3_2.12", version: "${project.analyticsVersion}")
+    api "org.apache.kafka:kafka-clients:${project.kafkaClientVersion}"
 }
 
 compileJava.onlyIf { !skipIntegrationTest }
 compileTestJava.onlyIf { !skipIntegrationTest }
 javadoc.onlyIf { !skipIntegrationTest }
+

integration-framework/src/main/java/org/apache/cassandra/sidecar/testing/MtlsTestHelper.java

@@ -22,6 +22,16 @@
 import java.util.Objects;
 import java.util.function.Consumer;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.sidecar.common.server.utils.SecondBoundConfiguration;
+import org.apache.cassandra.sidecar.config.KeyStoreConfiguration;
+import org.apache.cassandra.sidecar.config.SidecarClientConfiguration;
+import org.apache.cassandra.sidecar.config.SslConfiguration;
+import org.apache.cassandra.sidecar.config.yaml.KeyStoreConfigurationImpl;
+import org.apache.cassandra.sidecar.config.yaml.SidecarClientConfigurationImpl;
+import org.apache.cassandra.sidecar.config.yaml.SslConfigurationImpl;
 import org.apache.cassandra.testing.utils.tls.CertificateBuilder;
 import org.apache.cassandra.testing.utils.tls.CertificateBundle;
 
@@ -30,6 +40,7 @@
  */
 public class MtlsTestHelper
 {
+    private static final Logger LOGGER = LoggerFactory.getLogger(MtlsTestHelper.class);
     public static final String PASSWORD_STRING = "cassandra";
     public static final char[] PASSWORD = PASSWORD_STRING.toCharArray();
     /**
@@ -142,4 +153,78 @@ public String serverKeyStoreType()
     {
         return "PKCS12";
     }
+
+    /**
+     * Creates SSL configuration with the specified keystore and shared truststore.
+     *
+     * @param keyStorePath the path to the keystore
+     * @param keyStorePassword the keystore password
+     * @param keyStoreType the keystore type
+     * @return SslConfiguration with the provided keystore and shared truststore
+     */
+    private SslConfiguration createSslConfiguration(String keyStorePath,
+                                                    String keyStorePassword,
+                                                    String keyStoreType)
+    {
+        KeyStoreConfiguration truststoreConfiguration =
+        new KeyStoreConfigurationImpl(trustStorePath(),
+                                      trustStorePassword(),
+                                      trustStoreType(),
+                                      SecondBoundConfiguration.parse("60s"));
+
+        KeyStoreConfiguration keyStoreConfiguration =
+        new KeyStoreConfigurationImpl(keyStorePath,
+                                      keyStorePassword,
+                                      keyStoreType,
+                                      SecondBoundConfiguration.parse("60s"));
+
+        return SslConfigurationImpl.builder()
+                                   .enabled(true)
+                                   .keystore(keyStoreConfiguration)
+                                   .truststore(truststoreConfiguration)
+                                   .build();
+    }
+
+    /**
+     * Creates SSL configuration for the Sidecar server with mTLS settings if enabled.
+     *
+     * @return SslConfiguration with server keystore/truststore, or null if mTLS is not enabled
+     */
+    public SslConfiguration createServerSslConfiguration()
+    {
+        if (!isEnabled())
+        {
+            LOGGER.info("Not enabling mTLS for testing. Set '{}' to 'true' if you would like mTLS enabled.",
+                        CASSANDRA_INTEGRATION_TEST_ENABLE_MTLS);
+            return null;
+        }
+
+        LOGGER.info("Enabling test mTLS certificate/keystore for server.");
+        return createSslConfiguration(serverKeyStorePath(),
+                                      serverKeyStorePassword(),
+                                      serverKeyStoreType());
+    }
+
+    /**
+     * Creates a SidecarClientConfiguration with mTLS settings if mTLS is enabled.
+     *
+     * @return a SidecarClientConfiguration with mTLS settings, or null if mTLS is not enabled
+     */
+    public SidecarClientConfiguration createSidecarClientConfiguration()
+    {
+        if (!isEnabled())
+        {
+            LOGGER.info("Not enabling mTLS for testing. Set '{}' to 'true' if you would like mTLS enabled.",
+                        CASSANDRA_INTEGRATION_TEST_ENABLE_MTLS);
+            return new SidecarClientConfigurationImpl(null);
+        }
+
+        LOGGER.info("Enabling test mTLS certificate/keystore for client.");
+        SslConfiguration clientSslConfiguration =
+        createSslConfiguration(clientKeyStorePath(),
+                              clientKeyStorePassword(),
+                              serverKeyStoreType());
+
+        return new SidecarClientConfigurationImpl(clientSslConfiguration);
+    }
 }

integration-framework/src/main/java/org/apache/cassandra/sidecar/testing/SharedClusterIntegrationTestBase.java

@@ -36,6 +36,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.function.BooleanSupplier;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.stream.Collectors;
@@ -88,18 +89,15 @@
 import org.apache.cassandra.sidecar.common.server.utils.SidecarVersionProvider;
 import org.apache.cassandra.sidecar.common.server.utils.ThrowableUtils;
 import org.apache.cassandra.sidecar.config.JmxConfiguration;
-import org.apache.cassandra.sidecar.config.KeyStoreConfiguration;
 import org.apache.cassandra.sidecar.config.S3ClientConfiguration;
 import org.apache.cassandra.sidecar.config.S3ProxyConfiguration;
 import org.apache.cassandra.sidecar.config.ServiceConfiguration;
 import org.apache.cassandra.sidecar.config.SidecarConfiguration;
 import org.apache.cassandra.sidecar.config.SslConfiguration;
-import org.apache.cassandra.sidecar.config.yaml.KeyStoreConfigurationImpl;
 import org.apache.cassandra.sidecar.config.yaml.S3ClientConfigurationImpl;
 import org.apache.cassandra.sidecar.config.yaml.SchemaKeyspaceConfigurationImpl;
-import org.apache.cassandra.sidecar.config.yaml.ServiceConfigurationImpl;
 import org.apache.cassandra.sidecar.config.yaml.SidecarConfigurationImpl;
-import org.apache.cassandra.sidecar.config.yaml.SslConfigurationImpl;
+import org.apache.cassandra.sidecar.config.yaml.TestServiceConfiguration;
 import org.apache.cassandra.sidecar.coordination.ClusterLease;
 import org.apache.cassandra.sidecar.lifecycle.InJvmDTestLifecycleProvider;
 import org.apache.cassandra.sidecar.lifecycle.LifecycleProvider;
@@ -116,7 +114,6 @@
 import org.apache.cassandra.testing.TestVersionSupplier;
 
 import static org.apache.cassandra.sidecar.config.yaml.S3ClientConfigurationImpl.DEFAULT_API_CALL_TIMEOUT;
-import static org.apache.cassandra.sidecar.testing.MtlsTestHelper.CASSANDRA_INTEGRATION_TEST_ENABLE_MTLS;
 import static org.apache.cassandra.testing.DriverTestUtils.buildContactPoints;
 import static org.apache.cassandra.testing.utils.IInstanceUtils.tryGetIntConfig;
 import static org.assertj.core.api.Assertions.assertThat;
@@ -440,6 +437,30 @@ protected void waitForSchemaReady(ServerWrapper serverWrapper, long timeout, Tim
         .isTrue();
     }
 
+    /**
+     * Polls a condition until it returns true or timeout is reached.
+     * Uses System.nanoTime() for accurate timing and Uninterruptibles for consistent sleep behavior.
+     *
+     * @param condition the condition to check
+     * @param timeoutSeconds maximum time to wait in seconds
+     * @param pollIntervalMillis interval between checks in milliseconds
+     * @throws AssertionError if timeout is reached before condition is met
+     */
+    protected void waitUntil(BooleanSupplier condition, long timeoutSeconds, long pollIntervalMillis)
+    {
+        long startTime = System.nanoTime();
+        long timeoutNanos = TimeUnit.SECONDS.toNanos(timeoutSeconds);
+
+        while (!condition.getAsBoolean())
+        {
+            if (System.nanoTime() - startTime > timeoutNanos)
+            {
+                throw new AssertionError("Condition not met within " + timeoutSeconds + " seconds");
+            }
+            Uninterruptibles.sleepUninterruptibly(pollIntervalMillis, TimeUnit.MILLISECONDS);
+        }
+    }
+
     /**
      * Stops the Sidecar service
      *
@@ -775,48 +796,23 @@ public LifecycleProvider lifecycleProvider()
         }
 
         public static SidecarConfigurationImpl.Builder defaultConfigurationBuilder(
-        MtlsTestHelper mtlsTestHelper, Function<SidecarConfigurationImpl.Builder, SidecarConfigurationImpl.Builder> configurationOverrides)
+                MtlsTestHelper mtlsTestHelper,
+                Function<SidecarConfigurationImpl.Builder, SidecarConfigurationImpl.Builder> configurationOverrides)
         {
-            ServiceConfiguration conf = ServiceConfigurationImpl.builder()
-                                                                .host("0.0.0.0") // binds to all interfaces, potential security issue if left running for long
-                                                                .port(0) // let the test find an available port
+            ServiceConfiguration conf = TestServiceConfiguration.builder()
                                                                 .schemaKeyspaceConfiguration(SchemaKeyspaceConfigurationImpl.builder()
                                                                                                                             .isEnabled(true)
                                                                                                                             .build())
                                                                 .build();
 
-
-            SslConfiguration sslConfiguration = null;
-            if (mtlsTestHelper.isEnabled())
-            {
-                LOGGER.info("Enabling test mTLS certificate/keystore.");
-
-                KeyStoreConfiguration truststoreConfiguration =
-                new KeyStoreConfigurationImpl(mtlsTestHelper.trustStorePath(),
-                                              mtlsTestHelper.trustStorePassword(),
-                                              mtlsTestHelper.trustStoreType(),
-                                              SecondBoundConfiguration.parse("60s"));
-
-                KeyStoreConfiguration keyStoreConfiguration =
-                new KeyStoreConfigurationImpl(mtlsTestHelper.serverKeyStorePath(),
-                                              mtlsTestHelper.serverKeyStorePassword(),
-                                              mtlsTestHelper.serverKeyStoreType(),
-                                              SecondBoundConfiguration.parse("60s"));
-
-                sslConfiguration = SslConfigurationImpl.builder()
-                                                       .enabled(true)
-                                                       .keystore(keyStoreConfiguration)
-                                                       .truststore(truststoreConfiguration)
-                                                       .build();
-            }
-            else
-            {
-                LOGGER.info("Not enabling mTLS for testing purposes. Set '{}' to 'true' if you would " +
-                            "like mTLS enabled.", CASSANDRA_INTEGRATION_TEST_ENABLE_MTLS);
-            }
-            S3ClientConfiguration s3ClientConfig = new S3ClientConfigurationImpl("s3-client", 4, SecondBoundConfiguration.parse("60s"),
-                                                                                 5242880, DEFAULT_API_CALL_TIMEOUT,
-                                                                                 buildTestS3ProxyConfig());
+            SslConfiguration sslConfiguration = mtlsTestHelper.createServerSslConfiguration();
+            S3ClientConfiguration s3ClientConfig =
+            new S3ClientConfigurationImpl("s3-client",
+                                          4,
+                                          SecondBoundConfiguration.parse("60s"),
+                                          5242880,
+                                          DEFAULT_API_CALL_TIMEOUT,
+                                          buildTestS3ProxyConfig());
 
             SidecarConfigurationImpl.Builder builder = SidecarConfigurationImpl.builder()
                                                                                .serviceConfiguration(conf)