jmckenzie-dev commented on code in PR #224:
URL:
https://github.com/apache/cassandra-analytics/pull/224#discussion_r3633452726
##########
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:
Ah. Does this mean we need to drop it?
##########
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:
Ah - I was confused about this comment being down here as I was looking at
it in isolation with the if check and thinking it was describing what the if
check did, not the choice of placement of said check outside the loop. Maybe
augment this comment with something like "We call this outside the loop because
it needs..." to make that more clear.
##########
cassandra-analytics-cdc/src/main/java/org/apache/cassandra/cdc/api/SchemaSupplier.java:
##########
@@ -21,13 +21,32 @@
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 all tables relevant to CDC processing.
+ * Returns ALL tables (CDC-enabled and CDC-disabled) so that the bridge's
Schema.instance
Review Comment:
Should move this javadoc down to the method below and update this since the
interface now can do either.
--
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]
