jmckenzie-dev commented on code in PR #224:
URL:
https://github.com/apache/cassandra-analytics/pull/224#discussion_r3624413279
##########
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");
Review Comment:
In this case, we're proceeding to update the schema we have cached to
effectively wipe any prior state in the "no cdc enabled" case. What do you
think about us adding this context to the log? Because my initial naive
assumption was "if we hit this point why don't we fast-fail out?" and then
realized "Ah yes. We need to have our schema reflect this".
Concretely, something like:
"No CDC-enabled tables found; no cdc updates will be processed until CDC is
enabled on a table's schema"?
##########
cassandra-analytics-cdc-sidecar/src/main/java/org/apache/cassandra/cdc/sidecar/SidecarCdc.java:
##########
@@ -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()
Review Comment:
Same thing here; the pattern of:
1. `schemaSupplier.getTables()`
2. `.filter(CqlTable::cdc)`
##########
cassandra-analytics-cdc-codec/src/main/java/org/apache/cassandra/cdc/schemastore/CachingSchemaStore.java:
##########
@@ -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())
Review Comment:
What happens in the case where we had CDC tables then the DDL was changed
and they were disabled? Wouldn't we skip the publish here so we never get the
memo that there's no more CDC enabled?
##########
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");
Review Comment:
Is this a warn condition or ERROR? Is there ever a non-error case where a
schema supplier would return 0 tables at all for the "getAllTables" shaped case?
##########
cassandra-analytics-cdc/src/test/java/org/apache/cassandra/cdc/CdcTests.java:
##########
@@ -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 "
Review Comment:
Do we normally have . continuations indented on the same line as their root?
I find that a little hard / surprising to read (i.e. it's not immediately clear
these are coupled / dependent operations from code structure, have to infer
from leading .).
Vs for example:
```java
assertThat(tablesPassedToBridge.get())
.as("Schema.instance must be updated with ALL tables (CDC
and non-CDC) so a "...
.containsExactlyInAnyOrder(cdcTable, nonCdcTable);
```
I won't push back if our project norm is . on the same line as the lines
above, but I personally pretty strongly prefer the indentation.
##########
cassandra-four-zero-types/src/main/java/org/apache/cassandra/bridge/CassandraSchema.java:
##########
@@ -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());
Review Comment:
Since this can noop presumably on a lot of cases, I think we should consider
logging in the `maybeUpdateSchema` call in the case where we have an update.
The prior logging here made sense as it was a dupe process; here it's just
telling us that we already knew a table existed but not much else.
##########
cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/KryoSerializationTests.java:
##########
@@ -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),
Review Comment:
Why don't we have the `cdc` flag as part of our equality comparison and
hashcode impl?
##########
cassandra-four-zero-types/src/main/java/org/apache/cassandra/bridge/CassandraSchema.java:
##########
@@ -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;
Review Comment:
Why the if / continue structure (which is arguably glorified modern-day
`goto`) instead of an if / else if / else structure?
##########
cassandra-analytics-cdc-codec/src/main/java/org/apache/cassandra/cdc/schemastore/CachingSchemaStore.java:
##########
@@ -160,8 +169,11 @@ protected CqlToAvroSchemaConverter schemaConverter()
private void publishSchemas()
{
schemaSupplier
- .getCdcEnabledTables()
- .thenAccept(refreshedCdcTables -> {
+ .getTables()
+ .thenAccept(allTables -> {
+ Set<CqlTable> refreshedCdcTables = allTables.stream()
Review Comment:
We have this pattern of filtering down to just the CDC enabled tables in
multiple places. What do you think about augmenting the `SchemaSupplier` to
provide both:
```java
CompletableFuture<Set<CqlTable>> getTables();
CompletableFuture<Set<CqlTable>> getCDCEnabledTables();
```
That'd tidy up a lot of this duplicated stream code and give us a single
inflection point to make modifications if this pattern needs to change (i.e.
`CqlTable::cdc` becomes something else, or needs more data to resolve, for
instance)
##########
cassandra-four-zero-types/src/main/java/org/apache/cassandra/bridge/CassandraSchema.java:
##########
Review Comment:
This confused me. The name `cdcEnabledTables` followed by:
```java
cdcEnabledTables.forEach((ks, tables) -> tables.forEach(table -> {
LOGGER.warn("Disabling CDC on table keyspace={} table={}", ks,
table);
CassandraSchema.disableCdc(schema, ks, table);
}));
```
Read that we were removing cdcEnabledTables from our schema.
Renaming this variable to `remainingCdcEnabledTables` or
`schemaCdcEnabledTables` plus a comment would help disambiguate this so it's
clear we're not unregistering cdc enabled tables there at the end of the loop.
The logic looks right but this naming threw me for a loop.
##########
cassandra-four-zero-types/src/main/java/org/apache/cassandra/bridge/CassandraSchema.java:
##########
@@ -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.
+ *
+ * <p>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<TableIdentifier> tables)
+ {
+ unregisterTables(Schema.instance, tables);
+ }
+
+ public static void unregisterTables(@NotNull Schema schema, @NotNull
Set<TableIdentifier> 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)
Review Comment:
I think we should rename this. `CassandraSchema#unregisterTable` implies
it's for just straight "unregistered your tables", vs. a name like
`CassandraSchema#unregisterNonCdcTables`, which is where we see the "Refusing
to unregistered CDC-enabled table" log message making a lot more sense.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
