diff --git a/CHANGES.txt b/CHANGES.txt index 811417a7e..afd381b8c 100644 --- a/CHANGES.txt +++ b/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) 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..35251619e 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 @@ -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 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) { 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..14bff3296 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,11 @@ 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 cdcEnabledTables = FutureUtils.get(schemaSupplier.getCDCEnabledTables()); + Optional rfOp = cdcEnabledTables.stream() + .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..574843241 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,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 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) -> { 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..0d86adea8 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 @@ -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> 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> 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> getCDCEnabledTables() + { + return getTables().thenApply(tables -> tables.stream() + .filter(CqlTable::cdc) + .collect(Collectors.toSet())); + } } 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..b4490cb0e 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; @@ -166,7 +178,12 @@ public void testMockedCdc(CassandraVersion version) final int maxRows = 5000; final int batchSize = 500; final int numBatches = maxRows / batchSize; - TestSchema testSchema = TestSchema.basicBuilder(bridge).build(); + // withCdc(true) is required so table.cdc() is true — SchemaSupplier.getTables() now + // returns all tables, and Cdc.refreshSchema() filters to CDC-enabled ones via + // CqlTable#cdc(), not via any pre-filtering by the supplier itself. Without this, + // cdcEnabledTables ends up empty, keyspaceSupplier() watches no keyspaces, and the + // CDC scanner never sees the mutations this test writes. + TestSchema testSchema = TestSchema.basicBuilder(bridge).withCdc(true).build(); CqlTable table = testSchema.buildTable(); bridge.buildSchema(table.createStatement(), @@ -266,6 +283,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 unregisterNonCdcTables(@NotNull Set tables) + { + delegate.unregisterNonCdcTables(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..73ed6b229 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; @@ -327,7 +351,7 @@ public static CqlField.CqlType unwrapIfFrozen(CqlField.CqlType type) @Override public int hashCode() { - return Objects.hash(keyspace, table, createStatement, fields, udts); + return Objects.hash(keyspace, table, createStatement, fields, udts, cdc); } @Override @@ -351,7 +375,8 @@ public boolean equals(Object other) && Objects.equals(this.table, that.table) && Objects.equals(this.createStatement, that.createStatement) && Objects.equals(this.fields, that.fields) - && Objects.equals(this.udts, that.udts); + && Objects.equals(this.udts, that.udts) + && this.cdc == that.cdc; } public static class Serializer extends com.esotericsoftware.kryo.Serializer @@ -383,7 +408,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 +432,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..23fd78746 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,35 @@ public void testCqlTable(CassandraBridge bridge) assertThat(deserialized).isEqualTo(table); } + @ParameterizedTest + @MethodSource("org.apache.cassandra.bridge.VersionRunner#bridges") + public void testCqlTableCdcFlagSurvivesSerialization(CassandraBridge bridge) + { + 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).isEqualTo(table); + 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..b52e247f3 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,25 @@ 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. (Named + * to make that CDC-enabled-table refusal unsurprising at the call site, rather than + * implying unconditional removal of whatever is passed in.) + * + * @param tables the tables to unregister + */ + public abstract void unregisterNonCdcTables(@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-five-zero-bridge/src/test/java/org/apache/cassandra/bridge/CassandraSchemaTests.java b/cassandra-five-zero-bridge/src/test/java/org/apache/cassandra/bridge/CassandraSchemaTests.java index 7b3c2db85..0a5f13506 100644 --- a/cassandra-five-zero-bridge/src/test/java/org/apache/cassandra/bridge/CassandraSchemaTests.java +++ b/cassandra-five-zero-bridge/src/test/java/org/apache/cassandra/bridge/CassandraSchemaTests.java @@ -45,6 +45,7 @@ public void testUpdateCdcSchema() .withPartitionKey("a", BRIDGE.bigint()) .withClusteringKey("b", BRIDGE.text()) .withColumn("c", BRIDGE.timeuuid()) + .withCdc(true) .build(); final CqlTable cqlTable1 = testSchema1.buildTable(); @@ -52,6 +53,7 @@ public void testUpdateCdcSchema() .withPartitionKey("pk", BRIDGE.uuid()) .withClusteringKey("ck", BRIDGE.aInt()) .withColumn("val", BRIDGE.blob()) + .withCdc(true) .build(); final CqlTable cqlTable2 = testSchema2.buildTable(); 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..67e25aae1 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 unregisterNonCdcTables(@NotNull Set tables) + { + CassandraSchema.unregisterNonCdcTables(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..7d39071ef 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; @@ -45,6 +46,7 @@ public void testUpdateCdcSchema() .withPartitionKey("a", BRIDGE.bigint()) .withClusteringKey("b", BRIDGE.text()) .withColumn("c", BRIDGE.timeuuid()) + .withCdc(true) .build(); final CqlTable cqlTable1 = testSchema1.buildTable(); @@ -52,6 +54,7 @@ public void testUpdateCdcSchema() .withPartitionKey("pk", BRIDGE.uuid()) .withClusteringKey("ck", BRIDGE.aInt()) .withColumn("val", BRIDGE.blob()) + .withCdc(true) .build(); final CqlTable cqlTable2 = testSchema2.buildTable(); @@ -90,4 +93,50 @@ public void testUpdateCdcSchema() assertThat(CassandraSchema.isCdcEnabled(schema, cqlTable1)).isFalse(); assertThat(CassandraSchema.isCdcEnabled(schema, cqlTable2)).isFalse(); } + + @Test + public void testUnregisterNonCdcTables() + { + 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.unregisterNonCdcTables(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.unregisterNonCdcTables(schema, ImmutableSet.of(nonCdcId)); + assertThat(CassandraSchema.has(schema, nonCdcTable.keyspace(), nonCdcTable.table())).isFalse(); + + // refuses to unregister a table that is currently CDC-enabled + CassandraSchema.unregisterNonCdcTables(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.unregisterNonCdcTables(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..bf90bbf02 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 @@ -20,6 +20,7 @@ package org.apache.cassandra.bridge; import java.util.ArrayList; +import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Objects; @@ -48,6 +49,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; @@ -278,7 +280,7 @@ public static void maybeUpdateSchema(Schema schema, return; } - LOGGER.info("Schema change detected, updating new table schema keyspace={} table={}", keyspace, cqlTable.table()); + LOGGER.info("Schema change detected, updating table schema keyspace={} table={} cdc={}", keyspace, cqlTable.table(), enableCdc); SchemaUpdater.updateTable(s, ks.get(), updatedTable); }); } @@ -296,52 +298,147 @@ public static void updateCdcSchema(@NotNull Schema schema, .collect(Collectors.joining(","))); } - Map> cdcEnabledTables = CassandraSchema.cdcEnabledTables(schema); + Set currentlyCdcEnabled = currentlyCdcEnabledTables(schema); + for (CqlTable table : cdcTables) { table.udts().forEach(udt -> CassandraTypesImplementation.INSTANCE.updateUDTs(table.keyspace(), udt)); UUID tableId = tableIdLookup.lookup(table.keyspace(), table.table()); - if (cdcEnabledTables.containsKey(table.keyspace()) && cdcEnabledTables.get(table.keyspace()).contains(table.table())) + boolean previouslyCdcEnabled = currentlyCdcEnabled.contains(TableIdentifier.of(table.keyspace(), table.table())); + if (previouslyCdcEnabled) { - // table has cdc enabled already, update schema if it has changed - LOGGER.debug("CDC already enabled 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()); - continue; + // maybeUpdateSchema logs on its own when it actually performs an update. + CassandraSchema.maybeUpdateSchema(schema, partitioner, table, tableId, table.cdc()); } - - if (CassandraSchema.has(schema, table)) + else if (CassandraSchema.has(schema, table)) + { + // table exists but wasn't tracked as cdc-enabled (e.g. a non-CDC table the + // caller included in cdcTables anyway) — update if schema changed. + CassandraSchema.maybeUpdateSchema(schema, partitioner, table, tableId, table.cdc()); + } + else { - // 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()); - continue; + // 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 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), + "ColumnFamilyStore not initialized in the schema"); + Preconditions.checkArgument(CassandraSchema.isCdcEnabled(schema, table), + "CDC not enabled for table: " + table.keyspace() + "." + table.table()); + } } + } + disableCdcOnStaleTables(schema, currentlyCdcEnabled, cdcTables); + } - // 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) + private static Set currentlyCdcEnabledTables(Schema schema) + { + return CassandraSchema.cdcEnabledTables(schema) + .entrySet() + .stream() + .flatMap(e -> e.getValue().stream().map(table -> TableIdentifier.of(e.getKey(), table))) + .collect(Collectors.toSet()); + } + + /** + * Disables CDC on every table in {@code currentlyCdcEnabled} that is not CDC-enabled in + * {@code cdcTables} (dropped, or CDC disabled in its CREATE TABLE). + */ + private static void disableCdcOnStaleTables(Schema schema, Set currentlyCdcEnabled, Set cdcTables) + { + Set stillCdcEnabled = cdcTables.stream() + .filter(CqlTable::cdc) + .map(t -> TableIdentifier.of(t.keyspace(), t.table())) + .collect(Collectors.toSet()); + Set stale = new HashSet<>(currentlyCdcEnabled); + stale.removeAll(stillCdcEnabled); + + stale.forEach(id -> { + LOGGER.warn("Disabling CDC on table keyspace={} table={}", id.keyspace(), id.table()); + CassandraSchema.disableCdc(schema, id.keyspace(), id.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#unregisterNonCdcTables} 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 unregisterNonCdcTables(@NotNull Set tables) + { + unregisterNonCdcTables(Schema.instance, tables); + } + + public static void unregisterNonCdcTables(@NotNull Schema schema, @NotNull Set tables) + { + for (TableIdentifier id : tables) + { + String keyspace = id.keyspace(); + String table = id.table(); + try { - // verify TableMetadata and ColumnFamilyStore initialized in Schema - 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), - "ColumnFamilyStore not initialized in the schema"); - Preconditions.checkArgument(CassandraSchema.isCdcEnabled(schema, table), - "CDC not enabled for table: " + table.keyspace() + "." + table.table()); + unregisterNonCdcTable(schema, keyspace, table); + } + catch (RuntimeException e) + { + // Don't let one bad table abort unregistration of the rest of the batch. + LOGGER.warn("Failed to unregister table keyspace={} table={}", keyspace, table, e); } } - // existing table no longer with cdc = true, so disable - cdcEnabledTables.forEach((ks, tables) -> tables.forEach(table -> { - LOGGER.warn("Disabling CDC on table keyspace={} table={}", ks, table); - CassandraSchema.disableCdc(schema, ks, table); - })); + } + + private static void unregisterNonCdcTable(@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) 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)