diff --git a/cassandra-analytics-cdc-codec/src/main/java/org/apache/cassandra/cdc/schemastore/CachingSchemaStore.java b/cassandra-analytics-cdc-codec/src/main/java/org/apache/cassandra/cdc/schemastore/CachingSchemaStore.java index 949819084..491ca8030 100644 --- a/cassandra-analytics-cdc-codec/src/main/java/org/apache/cassandra/cdc/schemastore/CachingSchemaStore.java +++ b/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 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 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 refreshedCdcTables = allTables.stream() + .filter(CqlTable::cdc) + .collect(Collectors.toSet()); for (CqlTable cqlTable : refreshedCdcTables) { TableIdentifier tableIdentifier = TableIdentifier.of(cqlTable.keyspace(), cqlTable.table()); diff --git a/cassandra-analytics-cdc-sidecar/src/main/java/org/apache/cassandra/cdc/sidecar/SidecarCdc.java b/cassandra-analytics-cdc-sidecar/src/main/java/org/apache/cassandra/cdc/sidecar/SidecarCdc.java index f7e0513cc..2aecf2ca7 100644 --- a/cassandra-analytics-cdc-sidecar/src/main/java/org/apache/cassandra/cdc/sidecar/SidecarCdc.java +++ b/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; not closed by * {@code SidecarCdc} or {@code SidecarCdcBuilder} @@ -92,11 +92,12 @@ public static SidecarCdcBuilder builder(@NotNull String jobId, public void initSchema() { - Set tables = FutureUtils.get(schemaSupplier.getCdcEnabledTables()); - Optional rfOp = tables.stream() - .map(CqlTable::replicationFactor) - .filter(rf -> rf.getOptions().containsKey(dc())) - .max(Comparator.comparingInt(rf -> rf.getOptions().get(dc()))); + Set allTables = FutureUtils.get(schemaSupplier.getTables()); + Optional 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()) { diff --git a/cassandra-analytics-cdc-sidecar/src/main/java/org/apache/cassandra/cdc/sidecar/SidecarCdcBuilder.java b/cassandra-analytics-cdc-sidecar/src/main/java/org/apache/cassandra/cdc/sidecar/SidecarCdcBuilder.java index 4027db2db..559edc2f9 100644 --- a/cassandra-analytics-cdc-sidecar/src/main/java/org/apache/cassandra/cdc/sidecar/SidecarCdcBuilder.java +++ b/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; diff --git a/cassandra-analytics-cdc/src/main/java/org/apache/cassandra/cdc/Cdc.java b/cassandra-analytics-cdc/src/main/java/org/apache/cassandra/cdc/Cdc.java index b3270c9fa..8ac9cb21e 100644 --- a/cassandra-analytics-cdc/src/main/java/org/apache/cassandra/cdc/Cdc.java +++ b/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 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) -> { diff --git a/cassandra-analytics-cdc/src/main/java/org/apache/cassandra/cdc/api/SchemaSupplier.java b/cassandra-analytics-cdc/src/main/java/org/apache/cassandra/cdc/api/SchemaSupplier.java index 0bd55a2f4..309c4dc5a 100644 --- a/cassandra-analytics-cdc/src/main/java/org/apache/cassandra/cdc/api/SchemaSupplier.java +++ b/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> getCdcEnabledTables(); + CompletableFuture> getTables(); } diff --git a/cassandra-analytics-cdc/src/test/java/org/apache/cassandra/cdc/CdcTests.java b/cassandra-analytics-cdc/src/test/java/org/apache/cassandra/cdc/CdcTests.java index b67fc2fda..cd21e9e9f 100644 --- a/cassandra-analytics-cdc/src/test/java/org/apache/cassandra/cdc/CdcTests.java +++ b/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 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> 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> tablesPassedToBridge; + + RecordingCdc(CdcBuilder builder, AtomicReference> 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> tablesPassedToBridge; + + RecordingCdcBridge(CdcBridge delegate, AtomicReference> tablesPassedToBridge) + { + this.delegate = delegate; + this.tablesPassedToBridge = tablesPassedToBridge; + } + + @Override + public void updateCdcSchema(@NotNull Set cdcTables, + @NotNull Partitioner partitioner, + @NotNull TableIdLookup tableIdLookup) + { + tablesPassedToBridge.set(cdcTables); + delegate.updateCdcSchema(cdcTables, partitioner, tableIdLookup); + } + + @Override + public void unregisterTables(@NotNull Set 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 listener, + @Nullable Long startTimestampMicros, + boolean readCommitLogHeader) + { + return delegate.readLog(log, tokenRange, markers, partitionId, stats, executor, listener, startTimestampMicros, readCommitLogHeader); + } + + @Override + public CdcStreamScanner openCdcStreamScanner(Collection 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) diff --git a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/CqlTable.java b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/CqlTable.java index d71e471c4..7f946f3cb 100644 --- a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/CqlTable.java +++ b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/CqlTable.java @@ -49,6 +49,7 @@ public class CqlTable implements Serializable private final String createStatement; private final List fields; private final Set udts; + private final boolean cdc; private final Map fieldsMap; private final Set columnsWithUdts; @@ -75,11 +76,24 @@ public CqlTable(@NotNull String keyspace, @NotNull List fields, @NotNull Set udts, int indexCount) + { + this(keyspace, table, createStatement, replicationFactor, fields, udts, indexCount, false); + } + + public CqlTable(@NotNull String keyspace, + @NotNull String table, + @NotNull String createStatement, + @NotNull ReplicationFactor replicationFactor, + @NotNull List fields, + @NotNull Set udts, + int indexCount, + boolean cdc) { this.keyspace = keyspace; this.table = table; this.createStatement = createStatement; this.replicationFactor = replicationFactor; + this.cdc = cdc; this.fields = fields.stream().sorted().collect(Collectors.toList()); this.fieldsMap = this.fields.stream().collect(Collectors.toMap(CqlField::name, Function.identity())); this.partitionKeys = this.fields.stream().filter(CqlField::isPartitionKey).sorted().collect(Collectors.toList()); @@ -241,6 +255,16 @@ public String createStatement() return createStatement; } + /** + * Returns true if CDC is enabled on this table. + * This flag is explicitly set at construction time (e.g. from the CREATE TABLE statement) + * and stored as a field — it is not derived from createStatement at runtime. + */ + public boolean cdc() + { + return cdc; + } + public int indexCount() { return indexCount; @@ -383,7 +407,8 @@ public CqlTable read(Kryo kryo, Input input, Class type) udts.add((CqlField.CqlUdt) CqlField.CqlType.read(input, cassandraTypes)); } int indexCount = input.readInt(); - return new CqlTable(keyspace, table, createStatement, replicationFactor, fields, udts, indexCount); + boolean cdc = input.readBoolean(); + return new CqlTable(keyspace, table, createStatement, replicationFactor, fields, udts, indexCount, cdc); } @Override @@ -406,6 +431,7 @@ public void write(Kryo kryo, Output output, CqlTable table) udt.write(output); } output.writeInt(table.indexCount()); + output.writeBoolean(table.cdc()); } } } diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/KryoSerializationTests.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/KryoSerializationTests.java index 62e76774b..6e14316b1 100644 --- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/KryoSerializationTests.java +++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/KryoSerializationTests.java @@ -21,6 +21,7 @@ import java.math.BigInteger; import java.util.Arrays; +import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -269,6 +270,38 @@ public void testCqlTable(CassandraBridge bridge) assertThat(deserialized).isEqualTo(table); } + @ParameterizedTest + @MethodSource("org.apache.cassandra.bridge.VersionRunner#bridges") + public void testCqlTableCdcFlagSurvivesSerialization(CassandraBridge bridge) + { + // CqlTable.equals()/hashCode() deliberately exclude the cdc field (see CqlTable.java), + // so a round-trip test must assert cdc() directly rather than rely on equals() — an + // equals()-based assertion would pass even if the Kryo serializer silently dropped or + // corrupted the cdc value. + List fields = ImmutableList.of(new CqlField(true, false, false, "a", bridge.bigint(), 0)); + ReplicationFactor replicationFactor = new ReplicationFactor(ReplicationFactor.ReplicationStrategy.NetworkTopologyStrategy, + ImmutableMap.of("DC1", 3, "DC2", 3)); + + for (boolean cdc : new boolean[]{ true, false }) + { + CqlTable table = new CqlTable("test_keyspace", + "test_table", + "create table test_keyspace.test_table (a bigint, primary key(a));", + replicationFactor, + fields, + Collections.emptySet(), + 0, + cdc); + + Output out = serialize(bridge.getVersion(), table); + CqlTable deserialized = deserialize(bridge.getVersion(), out, CqlTable.class); + assertThat(deserialized).isNotNull(); + assertThat(deserialized.cdc()) + .as("cdc=%s must survive Kryo serialization round-trip", cdc) + .isEqualTo(cdc); + } + } + @ParameterizedTest @MethodSource("org.apache.cassandra.bridge.VersionRunner#bridges") public void testCassandraInstance(CassandraBridge bridge) diff --git a/cassandra-bridge/src/main/java/org/apache/cassandra/bridge/CdcBridge.java b/cassandra-bridge/src/main/java/org/apache/cassandra/bridge/CdcBridge.java index d233808b7..74c51cb09 100644 --- a/cassandra-bridge/src/main/java/org/apache/cassandra/bridge/CdcBridge.java +++ b/cassandra-bridge/src/main/java/org/apache/cassandra/bridge/CdcBridge.java @@ -42,6 +42,7 @@ import org.apache.cassandra.spark.data.CqlTable; import org.apache.cassandra.spark.data.partitioner.Partitioner; import org.apache.cassandra.spark.utils.AsyncExecutor; +import org.apache.cassandra.spark.utils.TableIdentifier; import org.apache.cassandra.spark.utils.TimeProvider; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; @@ -74,6 +75,23 @@ public abstract void updateCdcSchema(@NotNull Set cdcTables, @NotNull Partitioner partitioner, @NotNull TableIdLookup tableIdLookup); + /** + * Removes tables from the bridge's {@code Schema.instance} that were previously registered + * (via {@link #updateCdcSchema}) but are no longer needed — e.g. a non-CDC table that used + * to share partition-key structure with a CDC-enabled table in the same keyspace, where a + * later schema change (the CDC table dropped, or either table's partition key altered) means + * it can no longer be co-located with a CDC-enabled table's update in the same commit-log + * {@code Mutation}. + * + *

Implementations must be idempotent (a no-op for a table that isn't currently + * registered) and must never remove a table that is currently CDC-enabled — the caller is + * responsible for only passing tables it has determined are safe to unregister, but this is + * a last-line defense against silently dropping schema for a table CDC still needs. + * + * @param tables the tables to unregister + */ + public abstract void unregisterTables(@NotNull Set tables); + public abstract CommitLogReader.Result readLog(@NotNull CommitLog log, @Nullable TokenRange tokenRange, @NotNull CommitLogMarkers markers, diff --git a/cassandra-bridge/src/testFixtures/java/org/apache/cassandra/spark/utils/test/TestSchema.java b/cassandra-bridge/src/testFixtures/java/org/apache/cassandra/spark/utils/test/TestSchema.java index e96847496..505ada883 100644 --- a/cassandra-bridge/src/testFixtures/java/org/apache/cassandra/spark/utils/test/TestSchema.java +++ b/cassandra-bridge/src/testFixtures/java/org/apache/cassandra/spark/utils/test/TestSchema.java @@ -498,7 +498,8 @@ public CqlTable buildTable() rf, allFields, udts, - 0); + 0, + withCdc); } public void writeSSTable(TemporaryDirectory directory, diff --git a/cassandra-four-zero-bridge/src/main/java/org/apache/cassandra/bridge/AbstractCdcBridgeImplementation.java b/cassandra-four-zero-bridge/src/main/java/org/apache/cassandra/bridge/AbstractCdcBridgeImplementation.java index 77f8f8814..e0cb0a729 100644 --- a/cassandra-four-zero-bridge/src/main/java/org/apache/cassandra/bridge/AbstractCdcBridgeImplementation.java +++ b/cassandra-four-zero-bridge/src/main/java/org/apache/cassandra/bridge/AbstractCdcBridgeImplementation.java @@ -70,6 +70,7 @@ import org.apache.cassandra.spark.data.partitioner.Partitioner; import org.apache.cassandra.spark.utils.AsyncExecutor; import org.apache.cassandra.spark.utils.ByteBufferUtils; +import org.apache.cassandra.spark.utils.TableIdentifier; import org.apache.cassandra.spark.utils.TimeProvider; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; @@ -108,6 +109,11 @@ public void updateCdcSchema(@NotNull Set cdcTables, @NotNull Partition CassandraSchema.updateCdcSchema(cdcTables, partitioner, tableIdLookup); } + public void unregisterTables(@NotNull Set tables) + { + CassandraSchema.unregisterTables(tables); + } + public CommitLogReader.Result readLog(@NotNull CommitLog log, @Nullable TokenRange tokenRange, @NotNull CommitLogMarkers markers, diff --git a/cassandra-four-zero-bridge/src/test/java/org/apache/cassandra/bridge/CassandraSchemaTests.java b/cassandra-four-zero-bridge/src/test/java/org/apache/cassandra/bridge/CassandraSchemaTests.java index 7b3c2db85..4c213d095 100644 --- a/cassandra-four-zero-bridge/src/test/java/org/apache/cassandra/bridge/CassandraSchemaTests.java +++ b/cassandra-four-zero-bridge/src/test/java/org/apache/cassandra/bridge/CassandraSchemaTests.java @@ -26,6 +26,7 @@ import org.apache.cassandra.schema.Schema; import org.apache.cassandra.spark.data.CqlTable; +import org.apache.cassandra.spark.utils.TableIdentifier; import org.apache.cassandra.spark.utils.test.TestSchema; import org.apache.cassandra.spark.data.partitioner.Partitioner; @@ -90,4 +91,50 @@ public void testUpdateCdcSchema() assertThat(CassandraSchema.isCdcEnabled(schema, cqlTable1)).isFalse(); assertThat(CassandraSchema.isCdcEnabled(schema, cqlTable2)).isFalse(); } + + @Test + public void testUnregisterTables() + { + Schema schema = Schema.instance; + + TestSchema nonCdcSchema = TestSchema.builder(BRIDGE) + .withPartitionKey("a", BRIDGE.uuid()) + .withColumn("b", BRIDGE.text()) + .build(); + CqlTable nonCdcTable = nonCdcSchema.buildTable(); + TableIdentifier nonCdcId = TableIdentifier.of(nonCdcTable.keyspace(), nonCdcTable.table()); + + TestSchema cdcSchema = TestSchema.builder(BRIDGE) + .withKeyspace(nonCdcTable.keyspace()) + .withPartitionKey("a", BRIDGE.uuid()) + .withColumn("b", BRIDGE.text()) + .withCdc(true) + .build(); + CqlTable cdcTable = cdcSchema.buildTable(); + TableIdentifier cdcId = TableIdentifier.of(cdcTable.keyspace(), cdcTable.table()); + + // register both tables (as if they'd been found to share partition-key structure) + CassandraSchema.updateCdcSchema(schema, ImmutableSet.of(nonCdcTable, cdcTable), Partitioner.Murmur3Partitioner, (keyspace, table) -> null); + assertThat(CassandraSchema.has(schema, nonCdcTable.keyspace(), nonCdcTable.table())).isTrue(); + assertThat(CassandraSchema.has(schema, cdcTable.keyspace(), cdcTable.table())).isTrue(); + + // a later refresh determines nonCdcTable is no longer at risk — unregister it + CassandraSchema.unregisterTables(schema, ImmutableSet.of(nonCdcId)); + assertThat(CassandraSchema.has(schema, nonCdcTable.keyspace(), nonCdcTable.table())).isFalse(); + // the CDC-enabled table must be completely unaffected + assertThat(CassandraSchema.has(schema, cdcTable.keyspace(), cdcTable.table())).isTrue(); + assertThat(CassandraSchema.isCdcEnabled(schema, cdcTable)).isTrue(); + + // idempotent: unregistering an already-unregistered table is a no-op, not an error + CassandraSchema.unregisterTables(schema, ImmutableSet.of(nonCdcId)); + assertThat(CassandraSchema.has(schema, nonCdcTable.keyspace(), nonCdcTable.table())).isFalse(); + + // refuses to unregister a table that is currently CDC-enabled + CassandraSchema.unregisterTables(schema, ImmutableSet.of(cdcId)); + assertThat(CassandraSchema.has(schema, cdcTable.keyspace(), cdcTable.table())).isTrue(); + assertThat(CassandraSchema.isCdcEnabled(schema, cdcTable)).isTrue(); + + // unregistering an unknown table (never registered) is a no-op, not an error + CassandraSchema.unregisterTables(schema, ImmutableSet.of(TableIdentifier.of("unknown_ks", "unknown_table"))); + } } diff --git a/cassandra-four-zero-types/src/main/java/org/apache/cassandra/bridge/CassandraSchema.java b/cassandra-four-zero-types/src/main/java/org/apache/cassandra/bridge/CassandraSchema.java index bd4f2c78c..11221fd7e 100644 --- a/cassandra-four-zero-types/src/main/java/org/apache/cassandra/bridge/CassandraSchema.java +++ b/cassandra-four-zero-types/src/main/java/org/apache/cassandra/bridge/CassandraSchema.java @@ -48,6 +48,7 @@ import org.apache.cassandra.spark.data.CqlTable; import org.apache.cassandra.spark.data.partitioner.Partitioner; import org.apache.cassandra.spark.reader.SchemaBuilder; +import org.apache.cassandra.spark.utils.TableIdentifier; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; @@ -304,31 +305,27 @@ public static void updateCdcSchema(@NotNull Schema schema, UUID tableId = tableIdLookup.lookup(table.keyspace(), table.table()); if (cdcEnabledTables.containsKey(table.keyspace()) && cdcEnabledTables.get(table.keyspace()).contains(table.table())) { - // table has cdc enabled already, update schema if it has changed - LOGGER.debug("CDC already enabled keyspace={} table={}", table.keyspace(), table.table()); + // table already registered in Schema.instance — update if schema changed + LOGGER.debug("Table already in Schema.instance, updating if changed keyspace={} table={}", table.keyspace(), table.table()); cdcEnabledTables.get(table.keyspace()).remove(table.table()); - CassandraSchema.maybeUpdateSchema(schema, partitioner, table, tableId, true); - Preconditions.checkArgument(CassandraSchema.isCdcEnabled(schema, table), - "CDC not enabled for table: " + table.keyspace() + "." + table.table()); + CassandraSchema.maybeUpdateSchema(schema, partitioner, table, tableId, table.cdc()); continue; } if (CassandraSchema.has(schema, table)) { - // update schema if changed for existing table - LOGGER.info("Enabling CDC on existing table keyspace={} table={}", table.keyspace(), table.table()); - CassandraSchema.maybeUpdateSchema(schema, partitioner, table, tableId, true); - Preconditions.checkArgument(CassandraSchema.isCdcEnabled(schema, table), - "CDC not enabled for table: " + table.keyspace() + "." + table.table()); + // table exists but not in cdcEnabledTables tracking — update schema if changed + LOGGER.debug("Updating existing table keyspace={} table={} cdc={}", table.keyspace(), table.table(), table.cdc()); + CassandraSchema.maybeUpdateSchema(schema, partitioner, table, tableId, table.cdc()); continue; } - // new table so initialize table with cdc = true - LOGGER.info("Adding new CDC enabled table keyspace={} table={}", table.keyspace(), table.table()); - new SchemaBuilder(table, partitioner, tableId, true); - if (tableId != null) + // new table — register with the CDC flag from the create statement + LOGGER.info("Registering new table keyspace={} table={} cdc={}", table.keyspace(), table.table(), table.cdc()); + new SchemaBuilder(table, partitioner, tableId, table.cdc()); + if (tableId != null && table.cdc()) { - // verify TableMetadata and ColumnFamilyStore initialized in Schema + // verify CDC-enabled tables are correctly initialized TableId tableIdAfter = TableId.fromUUID(tableId); Preconditions.checkNotNull(schema.getTableMetadata(tableIdAfter), "Table not initialized in the schema"); Preconditions.checkArgument(Objects.requireNonNull(schema.getKeyspaceInstance(table.keyspace())).hasColumnFamilyStore(tableIdAfter), @@ -337,13 +334,92 @@ public static void updateCdcSchema(@NotNull Schema schema, "CDC not enabled for table: " + table.keyspace() + "." + table.table()); } } - // existing table no longer with cdc = true, so disable + // tables previously with cdc=true that are no longer present → disable CDC cdcEnabledTables.forEach((ks, tables) -> tables.forEach(table -> { LOGGER.warn("Disabling CDC on table keyspace={} table={}", ks, table); CassandraSchema.disableCdc(schema, ks, table); })); } + /** + * Removes tables from {@code Schema.instance} that were previously registered via + * {@link #updateCdcSchema} but are no longer needed — e.g. a non-CDC table that no longer + * shares partition-key structure with any CDC-enabled table in its keyspace after a schema + * change. See {@code CdcBridge#unregisterTables} for the full rationale. + * + *

Idempotent: a table not currently registered is silently skipped. Refuses (skips, with + * a warning) to unregister any table that is currently CDC-enabled — the caller is + * responsible for only requesting removal of tables it has determined are safe, but this is + * a last-line defense against silently dropping schema CDC still needs. + * + * @param tables the tables to unregister + */ + public static void unregisterTables(@NotNull Set tables) + { + unregisterTables(Schema.instance, tables); + } + + public static void unregisterTables(@NotNull Schema schema, @NotNull Set tables) + { + for (TableIdentifier id : tables) + { + String keyspace = id.keyspace(); + String table = id.table(); + try + { + unregisterTable(schema, keyspace, table); + } + catch (RuntimeException e) + { + // Never let one misbehaving table abort unregistration of the rest of the + // batch — the caller (CassandraClusterSchemaMonitor) treats this call as + // all-or-nothing for bookkeeping purposes, so a partial failure here must not + // prevent the other (unrelated) tables in the same batch from being cleaned up. + LOGGER.warn("Failed to unregister table keyspace={} table={}", keyspace, table, e); + } + } + } + + private static void unregisterTable(@NotNull Schema schema, @NotNull String keyspace, @NotNull String table) + { + Optional tableMetadata = getTable(schema, keyspace, table); + if (!tableMetadata.isPresent()) + { + // already unregistered (or never was) — nothing to do + return; + } + + if (tableMetadata.get().params.cdc) + { + LOGGER.warn("Refusing to unregister CDC-enabled table keyspace={} table={}", keyspace, table); + return; + } + + update(s -> { + Optional ks = getKeyspaceMetadata(s, keyspace); + Optional tableOpt = getTable(s, keyspace, table); + if (!ks.isPresent() || !tableOpt.isPresent()) + { + // unregistered by a concurrent call, or keyspace itself is gone + return; + } + if (tableOpt.get().params.cdc) + { + // became CDC-enabled since the check above (race with a concurrent + // updateCdcSchema) — do not remove it + return; + } + + LOGGER.info("Unregistering table no longer at risk of a batch with a CDC-enabled table keyspace={} table={}", keyspace, table); + // Only remove the table's schema metadata (so it goes back to throwing + // UnknownTableException on deserialization) — this bridge never performs real + // writes/compactions on the tables it mirrors, so a full Keyspace.dropCf() would + // pull in unrelated production machinery (e.g. lazily initializing + // CompactionManager's thread pools) with no corresponding benefit here. + SchemaUpdater.load(s, ks.get().withSwapped(ks.get().tables.without(table))); + }); + } + public static void enableCdc(Schema schema, CqlTable cqlTable) { enableCdc(schema, cqlTable.keyspace(), cqlTable.table()); diff --git a/cassandra-four-zero-types/src/main/java/org/apache/cassandra/spark/reader/SchemaBuilder.java b/cassandra-four-zero-types/src/main/java/org/apache/cassandra/spark/reader/SchemaBuilder.java index 5d0493a1f..b565565b2 100644 --- a/cassandra-four-zero-types/src/main/java/org/apache/cassandra/spark/reader/SchemaBuilder.java +++ b/cassandra-four-zero-types/src/main/java/org/apache/cassandra/spark/reader/SchemaBuilder.java @@ -85,6 +85,7 @@ public class SchemaBuilder private final ReplicationFactor replicationFactor; private final CassandraTypes cassandraTypes; private final int indexCount; + private final boolean enableCdc; public SchemaBuilder(CqlTable table, Partitioner partitioner, boolean enableCdc) { @@ -93,7 +94,7 @@ public SchemaBuilder(CqlTable table, Partitioner partitioner, boolean enableCdc) public SchemaBuilder(CqlTable table, Partitioner partitioner) { - this(table, partitioner, null, false); + this(table, partitioner, null, table.cdc()); } public SchemaBuilder(CqlTable table, Partitioner partitioner, UUID tableId, boolean enableCdc) @@ -137,6 +138,7 @@ public SchemaBuilder(String createStmt, this.replicationFactor = replicationFactor; this.cassandraTypes = new CassandraTypesImplementation(); this.indexCount = indexCount; + this.enableCdc = enableCdc; Pair updated = CassandraSchema.apply(schema -> updateSchema(schema, @@ -477,7 +479,8 @@ public CqlTable build() replicationFactor, fields, new HashSet<>(udts.values()), - indexCount); + indexCount, + enableCdc); } private Map buildsUdts(KeyspaceMetadata keyspaceMetadata)