feat - kafka support

.github/workflows/backfill-ci.yaml

@@ -30,6 +30,8 @@ jobs:
         env:
           DSE_REPO_USERNAME: ${{ secrets.DSE_REPO_USERNAME }}
           DSE_REPO_PASSWORD: ${{ secrets.DSE_REPO_PASSWORD }}
+          MAVEN_OPTS: "-Xmx2g -XX:MaxMetaspaceSize=512m"  # PHASE 1: Limit JVM memory
+          GRADLE_OPTS: "-Xmx2g -Dorg.gradle.daemon=false"  # PHASE 1: Limit Gradle memory
         run: |
           ./gradlew -PdseRepoUsername=$DSE_REPO_USERNAME -PdseRepoPassword=$DSE_REPO_PASSWORD \
           backfill-cli:build
@@ -72,6 +74,8 @@ jobs:
         env:
           DSE_REPO_USERNAME: ${{ secrets.DSE_REPO_USERNAME }}
           DSE_REPO_PASSWORD: ${{ secrets.DSE_REPO_PASSWORD }}
+          MAVEN_OPTS: "-Xmx2g -XX:MaxMetaspaceSize=512m"  # PHASE 1: Limit JVM memory
+          GRADLE_OPTS: "-Xmx2g -Dorg.gradle.daemon=false"  # PHASE 1: Limit Gradle memory
         run: |
           set -e
           PREV_IFS=$IFS
@@ -80,9 +84,58 @@ jobs:
           IFS=$PREV_IFS
           PULSAR_IMAGE=${PULSAR_FULL_IMAGE[0]}
           PULSAR_IMAGE_TAG=${PULSAR_FULL_IMAGE[1]}
-          
+
           ./gradlew -Pdse4 -PdseRepoUsername=$DSE_REPO_USERNAME -PdseRepoPassword=$DSE_REPO_PASSWORD \
+          -Papi.version=1.43 \
           -PtestPulsarImage=$PULSAR_IMAGE \
           -PtestPulsarImageTag=$PULSAR_IMAGE_TAG \
           -PcassandraFamily=${{ matrix.cassandraFamily }} \
           backfill-cli:e2eTest
+
+  test-kafka:
+    needs: build
+    name: Test Backfill CLI (Kafka)
+    runs-on: ubuntu-latest
+    timeout-minutes: 90
+    strategy:
+      fail-fast: false
+      matrix:
+        jdk: ['11']
+        kafkaImage: ['confluentinc/cp-kafka:7.8.8', 'confluentinc/cp-kafka:7.9.7', 'confluentinc/cp-kafka:8.1.3']
+        cassandraFamily: ['c3', 'c4', 'dse4']
+    steps:
+      - uses: actions/checkout@v6
+      - name: Set up JDK ${{ matrix.jdk }}
+        uses: actions/setup-java@v5
+        with:
+          java-version: ${{ matrix.jdk }}
+          distribution: 'adopt'
+
+      - name: Get project version
+        uses: HardNorth/github-version-generate@v1.4.0
+        with:
+          version-source: file
+          version-file: gradle.properties
+          version-file-extraction-pattern: '(?<=version=).+'
+
+      - name: Cache Docker layers
+        uses: actions/cache@v5
+        with:
+          path: /tmp/.buildx-cache
+          key: ${{ runner.os }}-buildx-${{ github.sha }}
+          restore-keys: |
+            ${{ runner.os }}-buildx-
+
+      - name: Test with Gradle
+        env:
+          DSE_REPO_USERNAME: ${{ secrets.DSE_REPO_USERNAME }}
+          DSE_REPO_PASSWORD: ${{ secrets.DSE_REPO_PASSWORD }}
+          MAVEN_OPTS: "-Xmx2g -XX:MaxMetaspaceSize=512m"
+          GRADLE_OPTS: "-Xmx2g -Dorg.gradle.daemon=false"
+        run: |
+          set -e
+          ./gradlew -Pdse4 -PdseRepoUsername=$DSE_REPO_USERNAME -PdseRepoPassword=$DSE_REPO_PASSWORD \
+          -Papi.version=1.43 \
+          -PkafkaImage=${{ matrix.kafkaImage }} \
+          -PcassandraFamily=${{ matrix.cassandraFamily }} \
+          backfill-cli:e2eTestKafka

.github/workflows/ci.yaml

@@ -32,15 +32,16 @@ jobs:
           DSE_REPO_PASSWORD: ${{ secrets.DSE_REPO_PASSWORD }}
         run: |
           ./gradlew -Pdse4 -PdseRepoUsername=$DSE_REPO_USERNAME -PdseRepoPassword=$DSE_REPO_PASSWORD \
-          build -x test -x backfill-cli:compileJava
+          build -x test -x backfill-cli:compileJava -x license -x licenseMain -x licenseTest
 
   test:
     needs: build
-    name: Test
+    name: Test Pulsar
     runs-on: ubuntu-latest
     timeout-minutes: 360
     strategy:
       fail-fast: false
+      #max-parallel: 10  # PHASE 1: Limit parallel test execution
       matrix:
         module: ['agent', 'agent-c3', 'agent-c4', 'agent-dse4', 'connector']
         jdk: ['11', '17']
@@ -72,6 +73,8 @@ jobs:
         env:
           DSE_REPO_USERNAME: ${{ secrets.DSE_REPO_USERNAME }}
           DSE_REPO_PASSWORD: ${{ secrets.DSE_REPO_PASSWORD }}
+          MAVEN_OPTS: "-Xmx2g -XX:MaxMetaspaceSize=512m"  # PHASE 1: Limit JVM memory
+          GRADLE_OPTS: "-Xmx2g -Dorg.gradle.daemon=false"  # PHASE 1: Limit Gradle memory, disable daemon
         run: |
           set -e
           PREV_IFS=$IFS
@@ -82,6 +85,64 @@ jobs:
           PULSAR_IMAGE_TAG=${PULSAR_FULL_IMAGE[1]}
 
           ./gradlew -Pdse4 -PdseRepoUsername=$DSE_REPO_USERNAME -PdseRepoPassword=$DSE_REPO_PASSWORD \
+          -Papi.version=1.43 \
           -PtestPulsarImage=$PULSAR_IMAGE \
           -PtestPulsarImageTag=$PULSAR_IMAGE_TAG \
           ${{ matrix.module }}:test
+
+  test-kafka:
+    needs: build
+    name: Test Kafka
+    runs-on: ubuntu-latest
+    timeout-minutes: 360
+    strategy:
+      fail-fast: false
+      matrix:
+        module: ['agent-c3', 'agent-c4', 'agent-dse4', 'connector-kafka']
+        jdk: ['11', '17']
+        kafkaImage: ['confluentinc/cp-kafka:7.8.8', 'confluentinc/cp-kafka:7.9.7', 'confluentinc/cp-kafka:8.1.3']
+    steps:
+      - uses: actions/checkout@v6
+      - name: Set up JDK ${{ matrix.jdk }}
+        uses: actions/setup-java@v5
+        with:
+          java-version: ${{ matrix.jdk }}
+          distribution: 'adopt'
+
+      - name: Get project version
+        uses: HardNorth/github-version-generate@v1.4.0
+        with:
+          version-source: file
+          version-file: gradle.properties
+          version-file-extraction-pattern: '(?<=version=).+'
+
+      - name: Cache Docker layers
+        uses: actions/cache@v5
+        with:
+          path: /tmp/.buildx-cache
+          key: ${{ runner.os }}-buildx-${{ github.sha }}
+          restore-keys: |
+            ${{ runner.os }}-buildx-
+
+      - name: Test with Gradle (Kafka)
+        env:
+          DSE_REPO_USERNAME: ${{ secrets.DSE_REPO_USERNAME }}
+          DSE_REPO_PASSWORD: ${{ secrets.DSE_REPO_PASSWORD }}
+          MAVEN_OPTS: "-Xmx2g -XX:MaxMetaspaceSize=512m"
+          GRADLE_OPTS: "-Xmx2g -Dorg.gradle.daemon=false"
+        run: |
+          set -e
+          PREV_IFS=$IFS
+          IFS=':'
+          read -ra KAFKA_FULL_IMAGE <<< "${{ matrix.kafkaImage }}"
+          IFS=$PREV_IFS
+          KAFKA_IMAGE=${KAFKA_FULL_IMAGE[0]}
+          KAFKA_IMAGE_TAG=${KAFKA_FULL_IMAGE[1]}
+
+          # -PkafkaTests includes the @Tag("kafka") integration tests (excluded by default).
+          ./gradlew -Pdse4 -PdseRepoUsername=$DSE_REPO_USERNAME -PdseRepoPassword=$DSE_REPO_PASSWORD \
+          -Papi.version=1.43 \
+          -PkafkaTests \
+          -PtestKafkaImage=$KAFKA_IMAGE \
+          -PtestKafkaImageTag=$KAFKA_IMAGE_TAG \
+          ${{ matrix.module }}:test

agent-c3/build.gradle

@@ -29,6 +29,8 @@ shadowJar {
     manifest {
         inheritFrom project.tasks.jar.manifest
     }
+    // Merge service provider files for SPI
+    mergeServiceFiles()
 }
 
 jar.enabled = false
@@ -37,8 +39,11 @@ assemble.dependsOn(shadowJar)
 dependencies {
     implementation project(':commons')
     implementation project(':agent')
-    implementation("org.apache.avro:avro:${avroVersion}")
+    implementation project(':messaging-api')
+    implementation project(':messaging-pulsar')
+    implementation project(':messaging-kafka')
 
+    implementation("org.apache.avro:avro:${avroVersion}")
     implementation("org.apache.pulsar:pulsar-client:${pulsarVersion}")
 
     compileOnly("org.apache.cassandra:cassandra-all:${cassandra3Version}")
@@ -68,6 +73,7 @@ test {
     useJUnitPlatform()
 
     environment 'PULSAR_IMAGE', testPulsarImage + ':' + testPulsarImageTag
+    environment 'KAFKA_IMAGE', testKafkaImage + ':' + testKafkaImageTag
     environment 'CASSANDRA_IMAGE', 'cassandra:' + cassandra3Version
 
     systemProperty "buildDir", buildDir

agent-c3/src/main/java/com/datastax/oss/cdc/agent/Agent.java

@@ -58,7 +58,9 @@ static void main(String agentArgs, Instrumentation inst) throws Exception {
 
     static void startCdcAgent(String agentArgs) throws Exception {
         log.info("Starting CDC agent, cdc_raw_directory={}", DatabaseDescriptor.getCDCLogLocation());
-        AgentConfig config = AgentConfig.create(AgentConfig.Platform.PULSAR, agentArgs);
+        // Platform.ALL: the agent is provider-agnostic and accepts both Pulsar and Kafka
+        // parameters; the active provider is selected at runtime via 'messagingProvider'.
+        AgentConfig config = AgentConfig.create(AgentConfig.Platform.ALL, agentArgs);
 
         // With C* 3.11, CL are immutable, we don't need to keep the last sent position.
         SegmentOffsetWriter segmentOffsetFileWriter = new SegmentOffsetDummyWriter(config.cdcWorkingDir);

agent-c3/src/main/java/com/datastax/oss/cdc/agent/PulsarMutationSender.java

@@ -56,7 +56,7 @@
 import java.util.UUID;
 
 @Slf4j
-public class PulsarMutationSender extends AbstractPulsarMutationSender<CFMetaData> {
+public class PulsarMutationSender extends AbstractMessagingMutationSender<CFMetaData> {
 
     private static final ImmutableMap<String, org.apache.avro.Schema> avroNativeTypes = ImmutableMap.<String, org.apache.avro.Schema>builder()
             .put(UTF8Type.instance.asCQL3Type().toString(), org.apache.avro.Schema.create(org.apache.avro.Schema.Type.STRING))
@@ -106,6 +106,12 @@ public org.apache.avro.Schema getNativeSchema(String cql3Type) {
      */
     @Override
     public boolean isSupported(final AbstractMutation<CFMetaData> mutation) {
+        // Check if metadata is null (table may have been dropped)
+        if (mutation.metadata == null) {
+            log.warn("Table metadata is null for mutation key={}, table may have been dropped, skipping mutation", mutation.key());
+            return false;
+        }
+
         if (!pkSchemas.containsKey(mutation.key())) {
             for (ColumnDefinition cm : mutation.metadata.primaryKeyColumns()) {
                 if (!avroNativeTypes.containsKey(cm.type.asCQL3Type().toString())) {

agent-c4/build.gradle

@@ -35,6 +35,8 @@ shadowJar {
     manifest {
         inheritFrom project.tasks.jar.manifest
     }
+    // Merge service provider files for SPI
+    mergeServiceFiles()
 }
 
 jar.enabled = true
@@ -43,6 +45,9 @@ assemble.dependsOn(shadowJar)
 dependencies {
     implementation project(':commons')
     implementation project(':agent')
+    implementation project(':messaging-api')
+    implementation project(':messaging-pulsar')
+    implementation project(':messaging-kafka')
 
     implementation("org.apache.avro:avro:${avroVersion}")
     implementation("commons-io:commons-io:${commonsIOVersion}") // Override transitive dependency version to fix vulnerability
@@ -77,6 +82,7 @@ test {
     useJUnitPlatform()
 
     environment 'PULSAR_IMAGE', testPulsarImage + ':' + testPulsarImageTag
+    environment 'KAFKA_IMAGE', testKafkaImage + ':' + testKafkaImageTag
     environment 'CASSANDRA_IMAGE', 'cassandra:' + cassandra4Version
 
     systemProperty "buildDir", buildDir

agent-c4/src/main/java/com/datastax/oss/cdc/agent/Agent.java

@@ -58,7 +58,9 @@ static void main(String agentArgs, Instrumentation inst) throws Exception {
 
     static void startCdcAgent(String agentArgs) throws Exception {
         log.info("Starting CDC agent, cdc_raw_directory={}", DatabaseDescriptor.getCDCLogLocation());
-        AgentConfig config = AgentConfig.create(AgentConfig.Platform.PULSAR, agentArgs);
+        // Platform.ALL: the agent is provider-agnostic and accepts both Pulsar and Kafka
+        // parameters; the active provider is selected at runtime via 'messagingProvider'.
+        AgentConfig config = AgentConfig.create(AgentConfig.Platform.ALL, agentArgs);
 
         SegmentOffsetFileWriter segmentOffsetFileWriter = new SegmentOffsetFileWriter(config.cdcWorkingDir);
         segmentOffsetFileWriter.loadOffsets();

agent-c4/src/main/java/com/datastax/oss/cdc/agent/PulsarMutationSender.java

@@ -56,7 +56,7 @@
 import java.util.UUID;
 
 @Slf4j
-public class PulsarMutationSender extends AbstractPulsarMutationSender<TableMetadata> {
+public class PulsarMutationSender extends AbstractMessagingMutationSender<TableMetadata> {
 
     private static final ImmutableMap<String, org.apache.avro.Schema> avroSchemaTypes = ImmutableMap.<String, org.apache.avro.Schema>builder()
             .put(UTF8Type.instance.asCQL3Type().toString(), org.apache.avro.Schema.create(org.apache.avro.Schema.Type.STRING))
@@ -110,6 +110,12 @@ public org.apache.avro.Schema getNativeSchema(String cql3Type) {
      */
     @Override
     public boolean isSupported(final AbstractMutation<TableMetadata> mutation) {
+        // Check if metadata is null (table may have been dropped)
+        if (mutation.metadata == null) {
+            log.warn("Table metadata is null for mutation key={}, table may have been dropped, skipping mutation", mutation.key());
+            return false;
+        }
+
         if (!pkSchemas.containsKey(mutation.key())) {
             for (ColumnMetadata cm : mutation.metadata.primaryKeyColumns()) {
                 if (!avroSchemaTypes.containsKey(cm.type.asCQL3Type().toString())) {

agent-c4/src/test/java/com/datastax/oss/cdc/agent/KafkaSingleNodeC4Tests.java

@@ -0,0 +1,49 @@
+/**
+ * Copyright DataStax, Inc 2021.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.cdc.agent;
+
+import com.datastax.oss.cdc.AgentTestUtil;
+import com.datastax.oss.cdc.KafkaSingleNodeTests;
+import com.datastax.testcontainers.cassandra.CassandraContainer;
+import lombok.extern.slf4j.Slf4j;
+import org.testcontainers.containers.Network;
+import org.testcontainers.utility.DockerImageName;
+
+import java.util.Optional;
+
+@Slf4j
+public class KafkaSingleNodeC4Tests extends KafkaSingleNodeTests {
+
+    public static final DockerImageName CASSANDRA_IMAGE = DockerImageName.parse(
+            Optional.ofNullable(System.getenv("CASSANDRA_IMAGE"))
+                    .orElse("cassandra:" + System.getProperty("cassandraVersion"))
+    ).asCompatibleSubstituteFor("cassandra");
+
+    public KafkaSingleNodeC4Tests() {
+        super(AgentTestUtil.Version.C4);
+    }
+
+    @Override
+    public CassandraContainer<?> createCassandraContainer(int nodeIndex, String kafkaBootstrapServers, Network testNetwork) {
+        return CassandraContainer.createCassandraContainerWithAgentKafka(
+                CASSANDRA_IMAGE, testNetwork, nodeIndex, "c4", kafkaBootstrapServers);
+    }
+
+    @Override
+    public int getSegmentSize() {
+        return 1024 * 1024;
+    }
+}

agent-dse4/README.md

@@ -1,11 +1,23 @@
-# DSE CDC agent for Apache Pulsar
+# DSE CDC agent for Apache Pulsar and Apache Kafka
+
+## Overview
+
+CDC agent for DataStax Enterprise 4.x with support for both Apache Pulsar and Apache Kafka.
 
 ## Build
 
     ./gradlew agent-dse4:shadowJar
 
-## Run
+## Run with Pulsar (Default)
+
+    export JVM_EXTRA_OPTS="-javaagent:agent-dse4/build/libs/agent-dse4-<version>-all.jar=pulsarServiceUrl=pulsar://pulsar:6650,cdcWorkingDir=/var/lib/cassandra/cdc"
+
+## Run with Kafka
+
+    export JVM_EXTRA_OPTS="-javaagent:agent-dse4/build/libs/agent-dse4-<version>-all.jar=messagingProvider=KAFKA,kafkaBootstrapServers=localhost:9092,cdcWorkingDir=/var/lib/cassandra/cdc"
+
+## Configuration
 
-    export JVM_EXTRA_OPTS="-javaagent:agent-dse4/build/libs/agent-dse4-<version>-SNAPSHOT-all.jar=pulsarServiceUrl=pulsar://pulsar:6650,cdcWorkingDir=/var/lib/cassandra/cdc"
+See [agent/README.md](../agent/README.md) for full configuration options.
 
 

agent-dse4/build.gradle

@@ -40,9 +40,15 @@ configurations {
 dependencies {
     custom project(':commons')
     custom project(':agent')
+    custom project(':messaging-api')
+    custom project(':messaging-pulsar')
+    custom project(':messaging-kafka')
 
     implementation project(':commons')
     implementation project(':agent')
+    implementation project(':messaging-api')
+    implementation project(':messaging-pulsar')
+    implementation project(':messaging-kafka')
 
     implementation("org.apache.avro:avro:${avroVersion}")
     implementation("${pulsarGroup}:pulsar-client:${pulsarVersion}")
@@ -84,6 +90,10 @@ shadowJar {
         inheritFrom project.tasks.jar.manifest
     }
     configurations = [project.configurations.custom]
+    // Merge service provider files for SPI
+    mergeServiceFiles()
+    // Exclude Netty native libraries; DSE provides its own bundled natives
+    exclude 'META-INF/native/*'
     // relocate AVRO because dse-db depends on avro 1.7.7
     relocate 'org.apache.avro', 'com.datastax.oss.cdc.avro'
 }