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

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

@@ -23,6 +23,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.Supplier;
@@ -83,8 +84,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 +107,10 @@ public void onConfigChange()
      */
     public void onSchemaChange()
     {
-        schemaSupplier.getCdcEnabledTables().thenAccept(refreshedCdcTables -> {
+        schemaSupplier.getTables().thenAccept(allTables -> {
+            Set<CqlTable> refreshedCdcTables = allTables.stream()
+                                                        .filter(CqlTable::cdc)
+                                                        .collect(Collectors.toSet());
             for (CqlTable cqlTable : refreshedCdcTables)
             {
                 TableIdentifier tableIdentifier = TableIdentifier.of(cqlTable.keyspace(), cqlTable.table());
@@ -118,11 +122,16 @@ public void onSchemaChange()
                     }
                     return value;
                 });
+            }
+            // publishSchemas() re-fetches and republishes every CDC table's schema, so it only
+            // needs to run once after the cache has been updated for all changed tables above —
+            // calling it per-table would redundantly re-fetch the schema 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.
+            // Remove any old schema entries for deleted tables
             List<TableIdentifier> refreshedTableIds = refreshedCdcTables
                                                       .stream()
                                                       .map(cqlTable -> TableIdentifier.of(cqlTable.keyspace(), cqlTable.table()))
@@ -160,8 +169,11 @@ protected CqlToAvroSchemaConverter schemaConverter()
     private void publishSchemas()
     {
         schemaSupplier
-        .getCdcEnabledTables()
-        .thenAccept(refreshedCdcTables -> {
+        .getTables()
+        .thenAccept(allTables -> {
+            Set<CqlTable> refreshedCdcTables = allTables.stream()
+                                                        .filter(CqlTable::cdc)
+                                                        .collect(Collectors.toSet());
             for (CqlTable cqlTable : refreshedCdcTables)
             {
                 TableIdentifier tableIdentifier = TableIdentifier.of(cqlTable.keyspace(), cqlTable.table());

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,12 @@ 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> allTables = FutureUtils.get(schemaSupplier.getTables());
+        Optional<ReplicationFactor> rfOp = allTables.stream()
+                                                    .filter(CqlTable::cdc)
+                                                    .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,33 @@ 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");
+                    LOGGER.warn("No tables returned from schema supplier");
                     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())
+                {
+                    LOGGER.warn("No CDC-enabled tables found");
+                }
+
+                // 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

@@ -25,9 +25,12 @@
 import org.apache.cassandra.spark.data.CqlTable;
 
 /**
- * Supplies all CDC enabled tables
+ * Supplies schema for all tables relevant to CDC processing.
+ * Returns 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.
  */
 public interface SchemaSupplier
 {
-    CompletableFuture<Set<CqlTable>> getCdcEnabledTables();
+    CompletableFuture<Set<CqlTable>> getTables();
 }

cassandra-analytics-cdc/src/test/java/org/apache/cassandra/cdc/CdcTests.java

@@ -25,6 +25,7 @@
 import java.nio.file.Path;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -56,17 +57,26 @@
 import org.apache.cassandra.bridge.CassandraVersion;
 import org.apache.cassandra.bridge.CdcBridge;
 import org.apache.cassandra.bridge.TokenRange;
+import org.apache.cassandra.cdc.api.CassandraSource;
 import org.apache.cassandra.cdc.api.CommitLog;
+import org.apache.cassandra.cdc.api.CommitLogInstance;
+import org.apache.cassandra.cdc.api.CommitLogMarkers;
 import org.apache.cassandra.cdc.api.CommitLogProvider;
+import org.apache.cassandra.cdc.api.CommitLogReader;
 import org.apache.cassandra.cdc.api.EventConsumer;
 import org.apache.cassandra.cdc.api.Marker;
+import org.apache.cassandra.cdc.api.Row;
 import org.apache.cassandra.cdc.api.SchemaSupplier;
 import org.apache.cassandra.cdc.api.StatePersister;
+import org.apache.cassandra.cdc.api.TableIdLookup;
 import org.apache.cassandra.cdc.msg.CdcEvent;
 import org.apache.cassandra.cdc.msg.Value;
+import org.apache.cassandra.cdc.scanner.CdcStreamScanner;
 import org.apache.cassandra.cdc.state.CdcState;
+import org.apache.cassandra.cdc.stats.ICdcStats;
 import org.apache.cassandra.cdc.test.CdcTestBase;
 import org.apache.cassandra.cdc.test.CdcTester;
+import org.apache.cassandra.db.commitlog.PartitionUpdateWrapper;
 import org.apache.cassandra.db.marshal.ByteBufferAccessor;
 import org.apache.cassandra.serializers.CollectionSerializer;
 import org.apache.cassandra.spark.data.CqlField;
@@ -77,6 +87,8 @@
 import org.apache.cassandra.spark.utils.AsyncExecutor;
 import org.apache.cassandra.spark.utils.ByteBufferUtils;
 import org.apache.cassandra.spark.utils.IOUtils;
+import org.apache.cassandra.spark.utils.TableIdentifier;
+import org.apache.cassandra.spark.utils.TimeProvider;
 import org.apache.cassandra.spark.utils.TimeUtils;
 import org.apache.cassandra.spark.utils.test.TestSchema;
 import org.apache.cassandra.transport.ProtocolVersion;
@@ -266,6 +278,171 @@ public List<CdcState> loadState(String jobId, int partitionId, @Nullable TokenRa
         }
     }
 
+    @ParameterizedTest
+    @MethodSource("org.apache.cassandra.cdc.test.TestVersionSupplier#testVersions")
+    public void testRefreshSchemaUpdatesBridgeWithAllTablesButFiltersCdcEnabledTables(CassandraVersion version)
+    {
+        // Regression test for the CDC batch-write bug: refreshSchema() must register EVERY
+        // table returned by the schema supplier (CDC-enabled and CDC-disabled) with the
+        // bridge's Schema.instance — not just the CDC-enabled ones — so that a commit log
+        // Mutation spanning both a CDC-enabled and a CDC-disabled table in the same keyspace
+        // (e.g. a BEGIN BATCH statement) can still be deserialized. Separately, only the
+        // CDC-enabled tables should end up in cdcEnabledTables, since that set drives
+        // publishing/replica-check decisions, not schema completeness.
+        TestSchema cdcSchema = TestSchema.builder(bridge)
+                                         .withPartitionKey("pk", bridge.uuid())
+                                         .withColumn("val", bridge.text())
+                                         .withCdc(true)
+                                         .build();
+        CqlTable cdcTable = cdcSchema.buildTable();
+
+        TestSchema nonCdcSchema = TestSchema.builder(bridge)
+                                            .withKeyspace(cdcSchema.keyspace)
+                                            .withPartitionKey("pk", bridge.uuid())
+                                            .withColumn("val", bridge.text())
+                                            .withCdc(false)
+                                            .build();
+        CqlTable nonCdcTable = nonCdcSchema.buildTable();
+
+        AtomicReference<Set<CqlTable>> tablesPassedToBridge = new AtomicReference<>();
+        SchemaSupplier schemaSupplier = () -> CompletableFuture.completedFuture(ImmutableSet.of(cdcTable, nonCdcTable));
+
+        try (RecordingCdc cdc = new RecordingCdc(Cdc.builder("test-refresh-schema", 0, event -> { }, schemaSupplier)
+                                                    .withExecutor(CdcTests.ASYNC_EXECUTOR)
+                                                    .withTableIdLookup(cdcBridge.internalTableIdLookup())
+                                                    .withCommitLogProvider(tokenRange -> Stream.empty())
+                                                    .withCdcOptions(cdcOptions),
+                                                    tablesPassedToBridge))
+        {
+            // start() flips isRunning to true and calls refreshSchema() synchronously (the
+            // schemaSupplier future is already completed, so the .handle()/.whenComplete()
+            // chain runs inline). scheduleRun()/scheduleMonitorSchema() are overridden to
+            // no-ops in RecordingCdc, so this only exercises the schema refresh path.
+            cdc.start();
+
+            assertThat(tablesPassedToBridge.get())
+            .as("Schema.instance must be updated with ALL tables (CDC and non-CDC) so a "
+              + "batch mutation spanning both can still deserialize")
+            .containsExactlyInAnyOrder(cdcTable, nonCdcTable);
+
+            assertThat(cdc.cdcEnabledTables)
+            .as("cdcEnabledTables (used for publishing/replica-check decisions) must contain "
+              + "only the CDC-enabled table, even though Schema.instance was given both")
+            .containsExactly(cdcTable);
+        }
+    }
+
+    /**
+     * Test-only {@link Cdc} subclass that intercepts {@link #cdcBridge()} to record exactly
+     * what table set gets passed to {@code CdcBridge.updateCdcSchema(...)} during
+     * {@code refreshSchema()}, while still delegating to the real bridge so Schema.instance is
+     * genuinely updated (not just observed). scheduleRun()/scheduleMonitorSchema() are
+     * overridden to no-ops so start() only exercises the schema refresh path, without kicking
+     * off an unrelated micro-batch read or a recurring schema-refresh loop.
+     */
+    private static final class RecordingCdc extends Cdc
+    {
+        private final AtomicReference<Set<CqlTable>> tablesPassedToBridge;
+
+        RecordingCdc(CdcBuilder builder, AtomicReference<Set<CqlTable>> tablesPassedToBridge)
+        {
+            super(builder);
+            this.tablesPassedToBridge = tablesPassedToBridge;
+        }
+
+        @Override
+        protected void scheduleRun(long delayMillis)
+        {
+            // no-op — this test only exercises schema refresh, not the micro-batch read loop
+        }
+
+        @Override
+        public void scheduleMonitorSchema()
+        {
+            // no-op — avoid recursively rescheduling refreshSchema() after start() calls it
+        }
+
+        @Override
+        protected CdcBridge cdcBridge()
+        {
+            CdcBridge real = super.cdcBridge();
+            return new RecordingCdcBridge(real, tablesPassedToBridge);
+        }
+    }
+
+    /**
+     * Delegates every call to the real {@link CdcBridge}, except {@code updateCdcSchema}, whose
+     * argument is captured before delegating.
+     */
+    private static final class RecordingCdcBridge extends CdcBridge
+    {
+        private final CdcBridge delegate;
+        private final AtomicReference<Set<CqlTable>> tablesPassedToBridge;
+
+        RecordingCdcBridge(CdcBridge delegate, AtomicReference<Set<CqlTable>> tablesPassedToBridge)
+        {
+            this.delegate = delegate;
+            this.tablesPassedToBridge = tablesPassedToBridge;
+        }
+
+        @Override
+        public void updateCdcSchema(@NotNull Set<CqlTable> cdcTables,
+                                    @NotNull Partitioner partitioner,
+                                    @NotNull TableIdLookup tableIdLookup)
+        {
+            tablesPassedToBridge.set(cdcTables);
+            delegate.updateCdcSchema(cdcTables, partitioner, tableIdLookup);
+        }
+
+        @Override
+        public void unregisterTables(@NotNull Set<TableIdentifier> tables)
+        {
+            delegate.unregisterTables(tables);
+        }
+
+        @Override
+        public CommitLogInstance createCommitLogInstance(Path path)
+        {
+            return delegate.createCommitLogInstance(path);
+        }
+
+        @Override
+        public TableIdLookup internalTableIdLookup()
+        {
+            return delegate.internalTableIdLookup();
+        }
+
+        @Override
+        public CommitLogReader.Result readLog(@NotNull CommitLog log,
+                                              @Nullable TokenRange tokenRange,
+                                              @NotNull CommitLogMarkers markers,
+                                              int partitionId,
+                                              @NotNull ICdcStats stats,
+                                              @Nullable AsyncExecutor executor,
+                                              @Nullable Consumer<Marker> listener,
+                                              @Nullable Long startTimestampMicros,
+                                              boolean readCommitLogHeader)
+        {
+            return delegate.readLog(log, tokenRange, markers, partitionId, stats, executor, listener, startTimestampMicros, readCommitLogHeader);
+        }
+
+        @Override
+        public CdcStreamScanner openCdcStreamScanner(Collection<PartitionUpdateWrapper> updates,
+                                                     @NotNull CdcState endState,
+                                                     Random random,
+                                                     CassandraSource cassandraSource,
+                                                     double traceSampleRate)
+        {
+            return delegate.openCdcStreamScanner(updates, endState, random, cassandraSource, traceSampleRate);
+        }
+
+        @Override
+        public void log(TimeProvider timeProvider, CqlTable cqlTable, CommitLogInstance log, Row row, long timestamp)
+        {
+            delegate.log(cqlTable, log, row, timestamp);
+        }
+    }
+
     @ParameterizedTest
     @MethodSource("org.apache.cassandra.cdc.test.TestVersionSupplier#testVersions")
     public void testSinglePartitionKey(CassandraVersion version)