Feature/cassandra 5 agent support

.github/workflows/backfill-ci.yaml

@@ -39,7 +39,7 @@ jobs:
       matrix:
         jdk: ['11'] # TODO: Enable java 17 tests https://issues.apache.org/jira/browse/CASSANDRA-16895
         pulsarImage: ['datastax/lunastreaming:2.10_3.4', 'apachepulsar/pulsar:2.10.3', 'apachepulsar/pulsar:2.11.0']
-        cassandraFamily: ['c3', 'c4', 'dse4']
+        cassandraFamily: ['c3', 'c4', 'c5', 'dse4']
     steps:
       - uses: actions/checkout@v3
       - name: Set up JDK ${{ matrix.jdk }}

.github/workflows/ci.yaml

@@ -37,7 +37,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        module: ['agent', 'agent-c3', 'agent-c4', 'agent-dse4', 'connector']
+        module: ['agent', 'agent-c3', 'agent-c4', 'agent-c5', 'agent-dse4', 'connector']
         jdk: ['11', '17']
         pulsarImage: ['datastax/lunastreaming:2.10_3.4', 'apachepulsar/pulsar:2.10.3', 'apachepulsar/pulsar:2.11.0']
     steps:

agent-c5/build.gradle

@@ -0,0 +1,161 @@
+plugins {
+    id 'java-library'
+    id 'java'
+    id 'application'
+    id "com.github.johnrengelman.shadow"
+    id 'com.palantir.docker'
+    id 'docker-compose'
+}
+
+// Cassandra 5.0 requires Java 17+ due to dependencies like Caffeine 3.1.8
+java {
+    toolchain {
+        languageVersion = JavaLanguageVersion.of(17)
+    }
+}
+
+sourceCompatibility = 17
+targetCompatibility = 17
+
+compileJava {
+    sourceCompatibility = 17
+    targetCompatibility = 17
+}
+
+application {
+    mainClass = "$mainClassName"
+}
+
+jar {
+    duplicatesStrategy = DuplicatesStrategy.INCLUDE
+    manifest {
+        attributes 'Premain-Class': "$mainClassName"
+    }
+    zip64=true
+}
+tasks.jar.dependsOn project(':commons').jar
+tasks.jar.dependsOn project(':agent').jar
+
+compileTestJava {
+    options.compilerArgs += '-parameters'
+}
+
+sourceSets {
+    // Make the compileOnly dependencies available when compiling/running tests
+    test.compileClasspath += configurations.compileClasspath
+    test.runtimeClasspath += configurations.compileClasspath
+}
+
+shadowJar {
+    manifest {
+        inheritFrom project.tasks.jar.manifest
+    }
+}
+
+jar.enabled = true
+assemble.dependsOn(shadowJar)
+
+dependencies {
+    implementation project(':commons')
+    implementation project(':agent')
+
+    implementation("org.apache.avro:avro:${avroVersion}")
+
+    implementation("${pulsarGroup}:pulsar-client:${pulsarVersion}")
+
+    compileOnly("org.apache.cassandra:cassandra-all:${cassandra5Version}")
+    testCompileOnly("org.apache.cassandra:cassandra-all:${cassandra5Version}")
+    testImplementation("org.apache.cassandra:cassandra-all:${cassandra5Version}") {
+        exclude group: 'io.netty'
+    }
+
+    testImplementation "com.datastax.oss:java-driver-core:${ossDriverVersion}"
+    testImplementation "com.datastax.oss:java-driver-query-builder:${ossDriverVersion}"
+
+    testImplementation "org.testcontainers:testcontainers:${testContainersVersion}"
+    testImplementation project(':testcontainers')
+
+    testRuntimeOnly "org.slf4j:slf4j-api:${slf4jVersion}"
+    testRuntimeOnly "ch.qos.logback:logback-classic:${logbackVersion}"
+}
+
+test {
+    // Add dependency on jar task, since it will be main target for testing
+    dependsOn shadowJar
+
+    // Rearrange test classpath, add compiled JAR instead of main classes
+    classpath = project.sourceSets.test.output + configurations.testRuntimeClasspath + files(shadowJar.archiveFile)
+
+    useJUnitPlatform()
+
+    // JVM arguments for Cassandra 5.0 Java 17+ compatibility
+    jvmArgs '--add-exports', 'java.base/jdk.internal.misc=ALL-UNNAMED',
+           '--add-exports', 'java.management.rmi/com.sun.jmx.remote.internal.rmi=ALL-UNNAMED',
+           '--add-exports', 'java.rmi/sun.rmi.registry=ALL-UNNAMED',
+           '--add-exports', 'java.rmi/sun.rmi.server=ALL-UNNAMED',
+           '--add-exports', 'java.sql/java.sql=ALL-UNNAMED',
+           '--add-exports', 'java.base/java.lang.ref=ALL-UNNAMED',
+           '--add-exports', 'jdk.unsupported/sun.misc=ALL-UNNAMED',
+           '--add-opens', 'java.base/java.lang.module=ALL-UNNAMED',
+           '--add-opens', 'java.base/jdk.internal.loader=ALL-UNNAMED',
+           '--add-opens', 'java.base/jdk.internal.ref=ALL-UNNAMED',
+           '--add-opens', 'java.base/jdk.internal.reflect=ALL-UNNAMED',
+           '--add-opens', 'java.base/jdk.internal.math=ALL-UNNAMED',
+           '--add-opens', 'java.base/jdk.internal.module=ALL-UNNAMED',
+           '--add-opens', 'java.base/jdk.internal.util.jar=ALL-UNNAMED',
+           '--add-opens', 'jdk.management/com.sun.management.internal=ALL-UNNAMED',
+           '--add-opens', 'java.base/sun.nio.ch=ALL-UNNAMED',
+           '--add-opens', 'java.base/java.io=ALL-UNNAMED',
+           '--add-opens', 'java.base/java.lang=ALL-UNNAMED',
+           '--add-opens', 'java.base/java.lang.reflect=ALL-UNNAMED',
+           '--add-opens', 'java.base/java.util=ALL-UNNAMED',
+           '--add-opens', 'java.base/java.nio=ALL-UNNAMED'
+
+    environment 'PULSAR_IMAGE', testPulsarImage + ':' + testPulsarImageTag
+    environment 'CASSANDRA_IMAGE', 'cassandra:' + cassandra5Version
+
+    systemProperty "buildDir", buildDir
+    systemProperty "projectVersion", project.version
+}
+
+dockerPrepare.dependsOn(jar, shadowJar)
+dockerPrepare.duplicatesStrategy = DuplicatesStrategy.INCLUDE
+docker {
+    name "${dockerRepo}cassandra:${cassandra5Version}-cdc"
+    files "$buildDir/libs/", "src/docker/"
+    buildArgs([BUILD_VERSION: project.version,
+               CASSANDRA_VERSION: cassandra5Version,
+               COMMITMOG_SYNC_PERIOD_IN_MS: commitlog_sync_period_in_ms,
+               CDC_TOTAL_SPACE_IN_MB: cdc_total_space_in_mb])
+}
+build.dependsOn(project.tasks['docker'])
+
+dockerCompose {
+    projectName = "test1"
+    useComposeFiles = ['../docker-compose.yaml']
+    environment.put 'AGENT_VERSION', 'c5'
+    environment.put 'PROJECT_VERSION', project.version
+    environment.put 'CASSANDRA_IMAGE', "${dockerRepo}cassandra:${cassandra5Version}-cdc"
+    environment.put 'PULSAR_IMAGE', "${dockerRepo}pulsar:${testPulsarImageTag}-csc"
+    waitForTcpPorts = false
+    startedServices = ['cassandra','pulsar']
+
+    stress {
+        projectName = "test1"
+        startedServices = ['cassandra-stress']
+        scale = ['cassandra-stress': 1]
+    }
+    stress2 {
+        projectName = "test1"
+        startedServices = ['cassandra-stress']
+        scale = ['cassandra-stress': 2]
+    }
+    prometheus {
+        projectName = "test1"
+        startedServices = ['prometheus','grafana']
+    }
+    elasticsearch {
+        projectName = "test1"
+        startedServices = ['elasticsearch']
+    }
+}

agent-c5/gradle.properties

@@ -0,0 +1,2 @@
+artifact=agent-c5
+mainClassName=com.datastax.oss.cdc.agent.Agent

agent-c5/src/docker/Dockerfile

@@ -0,0 +1,22 @@
+ARG CASSANDRA_VERSION
+FROM cassandra:${CASSANDRA_VERSION}
+ARG BUILD_VERSION
+ARG COMMITMOG_SYNC_PERIOD_IN_MS
+ARG CDC_TOTAL_SPACE_IN_MB
+
+RUN sed -i 's/cdc_enabled: false/cdc_enabled: true/g' /etc/cassandra/cassandra.yaml
+RUN sed -i "s/commitlog_sync_period_in_ms: 10000/commitlog_sync_period_in_ms: $COMMITMOG_SYNC_PERIOD_IN_MS/g" /etc/cassandra/cassandra.yaml
+RUN echo "cdc_total_space_in_mb: ${CDC_TOTAL_SPACE_IN_MB}" >> /etc/cassandra/cassandra.yaml
+
+COPY agent-c5-${BUILD_VERSION}-all.jar /
+
+# Add cassandra-stress config
+COPY table1.yaml /
+
+# Add the prometheus exporter
+ADD https://repo1.maven.org/maven2/io/prometheus/jmx/jmx_prometheus_javaagent/0.16.1/jmx_prometheus_javaagent-0.16.1.jar /
+RUN chmod a+rx /jmx_prometheus_javaagent-0.16.1.jar
+COPY jmx_prometheus_exporter.yaml /
+
+# Add cassandra tools in the PATH
+ENV PATH="${PATH}:/opt/cassandra/tools/bin"

agent-c5/src/docker/jmx_prometheus_exporter.yaml

@@ -0,0 +1,83 @@
+lowercaseOutputName: true
+lowercaseOutputLabelNames: true
+whitelistObjectNames: [
+    "org.apache.cassandra.metrics:type=ColumnFamily,name=RangeLatency,*",
+    "org.apache.cassandra.metrics:type=ColumnFamily,name=LiveSSTableCount,*",
+    "org.apache.cassandra.metrics:type=ColumnFamily,name=SSTablesPerReadHistogram,*",
+    "org.apache.cassandra.metrics:type=ColumnFamily,name=SpeculativeRetries,*",
+    "org.apache.cassandra.metrics:type=ColumnFamily,name=MemtableOnHeapSize,*",
+    "org.apache.cassandra.metrics:type=ColumnFamily,name=MemtableSwitchCount,*",
+    "org.apache.cassandra.metrics:type=ColumnFamily,name=MemtableLiveDataSize,*",
+    "org.apache.cassandra.metrics:type=ColumnFamily,name=MemtableColumnsCount,*",
+    "org.apache.cassandra.metrics:type=ColumnFamily,name=MemtableOffHeapSize,*",
+    "org.apache.cassandra.metrics:type=ColumnFamily,name=BloomFilterFalsePositives,*",
+    "org.apache.cassandra.metrics:type=ColumnFamily,name=BloomFilterFalseRatio,*",
+    "org.apache.cassandra.metrics:type=ColumnFamily,name=BloomFilterDiskSpaceUsed,*",
+    "org.apache.cassandra.metrics:type=ColumnFamily,name=BloomFilterOffHeapMemoryUsed,*",
+    "org.apache.cassandra.metrics:type=ColumnFamily,name=SnapshotsSize,*",
+    "org.apache.cassandra.metrics:type=ColumnFamily,name=TotalDiskSpaceUsed,*",
+    "org.apache.cassandra.metrics:type=CQL,name=RegularStatementsExecuted,*",
+    "org.apache.cassandra.metrics:type=CQL,name=PreparedStatementsExecuted,*",
+    "org.apache.cassandra.metrics:type=Compaction,name=PendingTasks,*",
+    "org.apache.cassandra.metrics:type=Compaction,name=CompletedTasks,*",
+    "org.apache.cassandra.metrics:type=Compaction,name=BytesCompacted,*",
+    "org.apache.cassandra.metrics:type=Compaction,name=TotalCompactionsCompleted,*",
+    "org.apache.cassandra.metrics:type=ClientRequest,name=Latency,*",
+    "org.apache.cassandra.metrics:type=ClientRequest,name=Unavailables,*",
+    "org.apache.cassandra.metrics:type=ClientRequest,name=Timeouts,*",
+    "org.apache.cassandra.metrics:type=ClientRequest,name=Failures,*",
+    "org.apache.cassandra.metrics:type=ClientRequest,name=Latency,*",
+    "org.apache.cassandra.metrics:type=Storage,name=Exceptions,*",
+    "org.apache.cassandra.metrics:type=Storage,name=TotalHints,*",
+    "org.apache.cassandra.metrics:type=Storage,name=TotalHintsInProgress,*",
+    "org.apache.cassandra.metrics:type=Storage,name=Load,*",
+    "org.apache.cassandra.metrics:type=Connection,name=TotalTimeouts,*",
+    "org.apache.cassandra.metrics:type=ThreadPools,name=CompletedTasks,*",
+    "org.apache.cassandra.metrics:type=ThreadPools,name=PendingTasks,*",
+    "org.apache.cassandra.metrics:type=ThreadPools,name=ActiveTasks,*",
+    "org.apache.cassandra.metrics:type=ThreadPools,name=TotalBlockedTasks,*",
+    "org.apache.cassandra.metrics:type=ThreadPools,name=CurrentlyBlockedTasks,*",
+    "org.apache.cassandra.metrics:type=DroppedMessage,name=Dropped,*",
+    "org.apache.cassandra.metrics:type=Cache,scope=KeyCache,name=HitRate,*",
+    "org.apache.cassandra.metrics:type=Cache,scope=KeyCache,name=Hits,*",
+    "org.apache.cassandra.metrics:type=Cache,scope=KeyCache,name=Requests,*",
+    "org.apache.cassandra.metrics:type=Cache,scope=KeyCache,name=Entries,*",
+    "org.apache.cassandra.metrics:type=Cache,scope=KeyCache,name=Size,*",
+    "org.apache.cassandra.metrics:type=Streaming,name=TotalIncomingBytes,*",
+    "org.apache.cassandra.metrics:type=Streaming,name=TotalOutgoingBytes,*",
+    "org.apache.cassandra.metrics:type=Client,name=connectedNativeClients,*",
+    "org.apache.cassandra.metrics:type=Client,name=connectedThriftClients,*",
+    "org.apache.cassandra.metrics:type=Table,name=WriteLatency,*",
+    "org.apache.cassandra.metrics:type=Table,name=ReadLatency,*",
+    "org.apache.cassandra.metrics:type=CdcAgent,*",
+    "org.apache.cassandra.metrics:type=CommitLog,*",
+    "org.apache.cassandra.net:type=FailureDetector,*",
+]
+#blacklistObjectNames: ["org.apache.cassandra.metrics:type=ColumnFamily,*"]
+rules:
+  - pattern: org.apache.cassandra.metrics<type=(Connection|Streaming), scope=(\S*), name=(\S*)><>(Count|Value)
+    name: cassandra_$1_$3
+    labels:
+      address: "$2"
+  - pattern: org.apache.cassandra.metrics<type=(ColumnFamily), name=(RangeLatency)><>(Mean)
+    name: cassandra_$1_$2_$3
+  - pattern: org.apache.cassandra.net<type=(FailureDetector)><>(DownEndpointCount)
+    name: cassandra_$1_$2
+  - pattern: org.apache.cassandra.metrics<type=(Keyspace), keyspace=(\S*), name=(\S*)><>(Count|Mean|95thPercentile)
+    name: cassandra_$1_$3_$4
+    labels:
+      "$1": "$2"
+  - pattern: org.apache.cassandra.metrics<type=(Table), keyspace=(\S*), scope=(\S*), name=(\S*)><>(Count|Mean|95thPercentile)
+    name: cassandra_$1_$4_$5
+    labels:
+      "keyspace": "$2"
+      "table": "$3"
+  - pattern: org.apache.cassandra.metrics<type=(ClientRequest), scope=(\S*), name=(\S*)><>(Count|Mean|95thPercentile)
+    name: cassandra_$1_$3_$4
+    labels:
+      "type": "$2"
+  - pattern: org.apache.cassandra.metrics<type=(\S*)(?:, ((?!scope)\S*)=(\S*))?(?:, scope=(\S*))?, name=(\S*)><>(Count|Value)
+    name: cassandra_$1_$5
+    labels:
+      "$1": "$4"
+      "$2": "$3"

agent-c5/src/docker/table1.yaml

@@ -0,0 +1,28 @@
+keyspace: ks1
+keyspace_definition: |
+  CREATE KEYSPACE ks1 WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '1'};
+#
+# Table name and create CQL
+#
+table: table1
+table_definition: |
+  CREATE TABLE table1 (a text, b text, PRIMARY KEY (a)) WITH cdc=true;
+
+columnspec:
+  - name: a
+    size: uniform(10..30)
+  - name: b
+    size: fixed(16)
+
+#
+# Specs for insert queries
+#
+insert:
+  partitions: fixed(1) # 1 partition per batch
+  batchtype: UNLOGGED # use unlogged batches
+  select: fixed(10)/10 # no chance of skipping a row when generating inserts
+
+queries:
+  read1:
+    cql: select * from table1 where a = ?
+    fields: samerow     # samerow or multirow (select arguments from the same row, or randomly from all rows in the partition)