CASSANALYTICS-182 : CDC batch-write mixing a CDC-enabled and CDC-disabled table drops the CDC table's mutation

CHANGES.txt

@@ -1,5 +1,6 @@
 0.5.0
 -----
+ * CDC batch-write mixing a CDC-enabled and CDC-disabled table drops the CDC table's mutation (CASSANALYTICS-182)
  * CdcState.ReplicaCountSerializer map-size overflow corrupts persisted CDC state (CASSANALYTICS-184)
  * SSTable-version-based bridge determination (CASSANALYTICS-24)
  * Upgrade sidecar version to 0.4.0 (CASSANALYTICS-176)

cassandra-analytics-cdc-codec/src/main/java/org/apache/cassandra/cdc/schemastore/CachingSchemaStore.java

@@ -83,8 +83,8 @@ public CachingSchemaStore(SchemaStoreStats schemaStoreStats,
     public void initialize()
     {
         LOGGER.info("Initializing CachingSchemaStore");
-        schemaSupplier.getCdcEnabledTables()
-                      .thenAccept(refreshedCdcTables -> {
+        schemaSupplier.getTables()
+                      .thenAccept(ignored -> {
                           loadPublisher();
                           publishSchemas();
                           LOGGER.info("CachingSchemaStore initialized");
@@ -106,7 +106,7 @@ public void onConfigChange()
      */
     public void onSchemaChange()
     {
-        schemaSupplier.getCdcEnabledTables().thenAccept(refreshedCdcTables -> {
+        schemaSupplier.getCDCEnabledTables().thenAccept(refreshedCdcTables -> {
             for (CqlTable cqlTable : refreshedCdcTables)
             {
                 TableIdentifier tableIdentifier = TableIdentifier.of(cqlTable.keyspace(), cqlTable.table());
@@ -118,11 +118,19 @@ public void onSchemaChange()
                     }
                     return value;
                 });
+            }
+            // We call publishSchemas() out here, after the loop, because it re-fetches and
+            // republishes every CDC table's schema in one pass — calling it inside the loop
+            // would redundantly re-fetch and republish every table N times over.
+            if (!refreshedCdcTables.isEmpty())
+            {
                 publishSchemas();
             }
-            // Remove any old schema entries for deleted tables, this operation can be done in the end as this is
-            // only for removing stale entries and no one is going to use these entries once the table is removed.
-            // This doesn't have to be an atomic operation.
+            else
+            {
+                LOGGER.warn("No CDC-enabled tables found; no schemas will be published until CDC is enabled on a table");
+            }
+            // Remove any old schema entries for deleted tables
             List<TableIdentifier> refreshedTableIds = refreshedCdcTables
                                                       .stream()
                                                       .map(cqlTable -> TableIdentifier.of(cqlTable.keyspace(), cqlTable.table()))
@@ -160,7 +168,7 @@ protected CqlToAvroSchemaConverter schemaConverter()
     private void publishSchemas()
     {
         schemaSupplier
-        .getCdcEnabledTables()
+        .getCDCEnabledTables()
         .thenAccept(refreshedCdcTables -> {
             for (CqlTable cqlTable : refreshedCdcTables)
             {

cassandra-analytics-cdc-sidecar/src/main/java/org/apache/cassandra/cdc/sidecar/SidecarCdc.java

@@ -62,7 +62,7 @@ protected SidecarCdc(@NotNull SidecarCdcBuilder builder)
      * @param cdcOptions             CDC processing options
      * @param clusterConfigProvider  provider for cluster configuration (e.g. datacenter, hosts)
      * @param eventConsumer          consumer that receives CDC change events
-     * @param schemaSupplier         supplier for CDC-enabled table schemas
+     * @param schemaSupplier         supplier for all table schemas (CDC-enabled and disabled); see {@link SchemaSupplier}
      * @param tokenRangeSupplier     supplier for the token ranges assigned to this partition
      * @param sidecarCdcClient       externally managed Sidecar HTTP client; <em>not</em> closed by
      *                               {@code SidecarCdc} or {@code SidecarCdcBuilder}
@@ -92,11 +92,11 @@ public static SidecarCdcBuilder builder(@NotNull String jobId,
 
     public void initSchema()
     {
-        Set<CqlTable> tables = FutureUtils.get(schemaSupplier.getCdcEnabledTables());
-        Optional<ReplicationFactor> rfOp = tables.stream()
-                                                 .map(CqlTable::replicationFactor)
-                                                 .filter(rf -> rf.getOptions().containsKey(dc()))
-                                                 .max(Comparator.comparingInt(rf -> rf.getOptions().get(dc())));
+        Set<CqlTable> cdcEnabledTables = FutureUtils.get(schemaSupplier.getCDCEnabledTables());
+        Optional<ReplicationFactor> rfOp = cdcEnabledTables.stream()
+                                                           .map(CqlTable::replicationFactor)
+                                                           .filter(rf -> rf.getOptions().containsKey(dc()))
+                                                           .max(Comparator.comparingInt(rf -> rf.getOptions().get(dc())));
 
         if (!rfOp.isPresent())
         {

cassandra-analytics-cdc-sidecar/src/main/java/org/apache/cassandra/cdc/sidecar/SidecarCdcBuilder.java

@@ -25,6 +25,7 @@
 import org.apache.cassandra.cdc.api.CdcOptions;
 import org.apache.cassandra.cdc.api.EventConsumer;
 import org.apache.cassandra.cdc.api.SchemaSupplier;
+import org.apache.cassandra.cdc.api.TableIdLookup;
 import org.apache.cassandra.cdc.api.TokenRangeSupplier;
 import org.apache.cassandra.cdc.stats.ICdcStats;
 import org.apache.cassandra.spark.utils.AsyncExecutor;
@@ -102,6 +103,13 @@ public SidecarCdcBuilder withExecutor(AsyncExecutor asyncExecutor)
         return withSidecarCdcCassandraClient(cassandraClient); // rebuild SidecarStatePersister with new AsyncExecutor
     }
 
+    @Override
+    public SidecarCdcBuilder withTableIdLookup(@NotNull TableIdLookup tableIdLookup)
+    {
+        super.withTableIdLookup(tableIdLookup);
+        return this;
+    }
+
     public SidecarCdcBuilder withSidecarCdcCassandraClient(SidecarCdcCassandraClient cassandraClient)
     {
         this.cassandraClient = cassandraClient;

cassandra-analytics-cdc/src/main/java/org/apache/cassandra/cdc/Cdc.java

@@ -415,22 +415,39 @@ protected void refreshSchema()
         try
         {
             schemaSupplier
-            .getCdcEnabledTables()
-            .handle((tables, throwable) -> {
+            .getTables()
+            .handle((allTables, throwable) -> {
                 if (throwable != null)
                 {
                     LOGGER.warn("Error refreshing schema", throwable);
                     return null;
                 }
-                this.cdcEnabledTables = tables;
-                if (tables == null || tables.isEmpty())
+                if (allTables == null || allTables.isEmpty())
                 {
-                    LOGGER.warn("No CQL enabled tables");
+                    // A brand-new cluster with no user tables yet is expected,
+                    // not a failure — refreshSchema() retries on its next scheduled run.
+                    LOGGER.warn("No tables returned from schema supplier; will retry on next scheduled schema refresh");
                     return null;
                 }
 
-                // update Schema instance with latest schema
-                cdcBridge().updateCdcSchema(tables, cdcOptions.partitioner(), tableIdLookup);
+                // Filter CDC-enabled tables for publishing decisions
+                Set<CqlTable> cdcTables = allTables.stream()
+                                                   .filter(CqlTable::cdc)
+                                                   .collect(Collectors.toSet());
+                this.cdcEnabledTables = cdcTables;
+                if (cdcTables.isEmpty())
+                {
+                    // Schema.instance is still updated below with allTables (which is non-empty
+                    // here), so this doesn't fast-fail — it just means no CDC updates will be
+                    // processed until CDC is enabled on at least one table's schema.
+                    LOGGER.warn("No CDC-enabled tables found; no cdc updates will be processed "
+                              + "until CDC is enabled on a table's schema");
+                }
+
+                // Update Schema.instance with ALL tables so deserialization never throws
+                // UnknownTableException for non-CDC tables co-located in batch mutations.
+                // Each table's CDC flag is set correctly from CqlTable.cdc().
+                cdcBridge().updateCdcSchema(allTables, cdcOptions.partitioner(), tableIdLookup);
                 return null;
             })
             .whenComplete((aVoid, throwable) -> {

cassandra-analytics-cdc/src/main/java/org/apache/cassandra/cdc/api/SchemaSupplier.java

@@ -21,13 +21,34 @@
 
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
+import java.util.stream.Collectors;
 
 import org.apache.cassandra.spark.data.CqlTable;
 
 /**
- * Supplies all CDC enabled tables
+ * Supplies schema for tables relevant to CDC processing.
  */
 public interface SchemaSupplier
 {
-    CompletableFuture<Set<CqlTable>> getCdcEnabledTables();
+    /**
+     * @return ALL tables (CDC-enabled and CDC-disabled) so that the bridge's Schema.instance
+     * is complete enough to deserialize any commit log mutation without UnknownTableException.
+     * Callers use {@link org.apache.cassandra.spark.data.CqlTable#cdc()} to filter for publishing.
+     */
+    CompletableFuture<Set<CqlTable>> getTables();
+
+    /**
+     * @return the subset of {@link #getTables()} that are CDC-enabled — i.e. what to actually
+     * publish/process, as opposed to the full set needed for schema completeness. A default
+     * method (rather than requiring implementations to filter themselves) so every caller that
+     * only cares about CDC-enabled tables shares one implementation of the
+     * {@code getTables().filter(CqlTable::cdc)} pattern, instead of repeating it at each call
+     * site.
+     */
+    default CompletableFuture<Set<CqlTable>> getCDCEnabledTables()
+    {
+        return getTables().thenApply(tables -> tables.stream()
+                                                      .filter(CqlTable::cdc)
+                                                      .collect(Collectors.toSet()));
+    }
 }