This is an automated email from the ASF dual-hosted git repository.
jyothsnakonisa pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra-sidecar.git
The following commit(s) were added to refs/heads/trunk by this push:
new 7972401c CASSSIDECAR-493 : CachingSchemaStore should be reloaded when
Schemastore kafka configs are changed (#377)
7972401c is described below
commit 7972401c6c4c0eb0f8968ba2544dc21486eccfd4
Author: Jyothsna konisa <[email protected]>
AuthorDate: Mon Aug 17 17:31:34 2026 -0700
CASSSIDECAR-493 : CachingSchemaStore should be reloaded when Schemastore
kafka configs are changed (#377)
Patch by Jyothsna Konisa; Reviewed by Saranya Krishnakumar for
CASSSIDECAR-493
---
CHANGES.txt | 1 +
.../cassandra/sidecar/cdc/CachingSchemaStore.java | 49 +++++++++++-------
.../sidecar/cdc/CachingSchemaStoreTest.java | 60 ++++++++++++++++++++++
3 files changed, 92 insertions(+), 18 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 080a74dd..3772e216 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,5 +1,6 @@
0.5.0
-----
+ * CachingSchemaStore should be reloaded when Schemastore kafka configs are
changed (CASSSIDECAR-493)
* Support retrying on HTTP 429 responses and fix duplicate immediate
execution in RequestExecutor retry scheduling (CASSSIDECAR-465)
* Add RFC 6902-inspired JSON Patch support to ConfigurationManager
(CASSSIDECAR-429)
* Implement job coordination for cluster-wide operations (CASSSIDECAR-377)
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/cdc/CachingSchemaStore.java
b/server/src/main/java/org/apache/cassandra/sidecar/cdc/CachingSchemaStore.java
index 2c812a34..18fef692 100644
---
a/server/src/main/java/org/apache/cassandra/sidecar/cdc/CachingSchemaStore.java
+++
b/server/src/main/java/org/apache/cassandra/sidecar/cdc/CachingSchemaStore.java
@@ -53,6 +53,7 @@ import org.apache.cassandra.spark.data.CqlTable;
import org.apache.cassandra.spark.utils.TableIdentifier;
import static
org.apache.cassandra.sidecar.server.SidecarServerEvents.ON_CDC_CACHE_WARMED_UP;
+import static
org.apache.cassandra.sidecar.server.SidecarServerEvents.ON_CDC_CONFIGURATION_CHANGED;
import static
org.apache.cassandra.sidecar.server.SidecarServerEvents.ON_SERVER_START;
import static
org.apache.cassandra.sidecar.server.SidecarServerEvents.ON_SIDECAR_SCHEMA_INITIALIZED;
@@ -149,19 +150,39 @@ public class CachingSchemaStore implements SchemaStore
return v;
});
}
- try
- {
- loadPublisher();
- publishSchemas();
- }
- catch (Exception e)
- {
- LOGGER.error("Failed to publish schemas to Kafka during
initialization, CDC will still start", e);
- }
+ reloadPublisherAndPublishSchemas("initialization");
+ });
+
+ // Retry on config fix, not just on schema change. Without this
listener, the only
+ // retry trigger was onSchemaChanged() (a CQL change) — unrelated
to the `configs`
+ // table, so a bad Schema Store config stayed broken on a
schema-stable cluster even
+ // after being corrected, until the next CQL change or a restart.
Mirrors how
+ // CdcPublisher already reacts to this same event to restart CDC
consumers.
+ eventBus.localConsumer(ON_CDC_CONFIGURATION_CHANGED.address(),
message -> {
+ LOGGER.info("Cdc configuration changed, reloading Schema Store
publisher");
+ reloadPublisherAndPublishSchemas("configuration change");
});
});
}
+ /**
+ * Rebuilds {@link #publisher} from the current Kafka config and
republishes every CDC
+ * table's merged schema. Failures are logged and swallowed — a broken
Schema Store config
+ * must never block CDC from starting/running.
+ */
+ private void reloadPublisherAndPublishSchemas(String trigger)
+ {
+ try
+ {
+ loadPublisher();
+ publishSchemas();
+ }
+ catch (Exception e)
+ {
+ LOGGER.error("Failed to publish schemas to Kafka during {}, CDC
will still start", trigger, e);
+ }
+ }
+
private void publishSchemas()
{
Set<CqlTable> refreshedCdcTables =
cassandraClusterSchemaMonitor.getCdcTables();
@@ -221,15 +242,7 @@ public class CachingSchemaStore implements SchemaStore
return v;
});
}
- try
- {
- loadPublisher();
- publishSchemas();
- }
- catch (Exception e)
- {
- LOGGER.error("Failed to publish schemas to Kafka, CDC will still
start", e);
- }
+ reloadPublisherAndPublishSchemas("schema change");
// 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.
diff --git
a/server/src/test/java/org/apache/cassandra/sidecar/cdc/CachingSchemaStoreTest.java
b/server/src/test/java/org/apache/cassandra/sidecar/cdc/CachingSchemaStoreTest.java
index 77746603..ecc1c2a2 100644
---
a/server/src/test/java/org/apache/cassandra/sidecar/cdc/CachingSchemaStoreTest.java
+++
b/server/src/test/java/org/apache/cassandra/sidecar/cdc/CachingSchemaStoreTest.java
@@ -23,7 +23,9 @@ import java.util.HashSet;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.EnumSource;
@@ -40,6 +42,7 @@ import org.apache.cassandra.cdc.api.TableIdLookup;
import org.apache.cassandra.cdc.avro.AvroSchemas;
import org.apache.cassandra.cdc.avro.CqlToAvroSchemaConverter;
import org.apache.cassandra.cdc.schemastore.SchemaStorePublisherFactory;
+import org.apache.cassandra.cdc.schemastore.TableSchemaPublisher;
import org.apache.cassandra.cdc.sidecar.SidecarCdcStats;
import org.apache.cassandra.sidecar.bridge.CassandraBridgeFactory;
import org.apache.cassandra.sidecar.db.TableHistoryDatabaseAccessor;
@@ -54,6 +57,10 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertSame;
+import static
org.apache.cassandra.sidecar.server.SidecarServerEvents.ON_CDC_CONFIGURATION_CHANGED;
+import static
org.apache.cassandra.sidecar.server.SidecarServerEvents.ON_SERVER_START;
+import static org.apache.cassandra.testing.utils.AssertionUtils.loopAssert;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatRuntimeException;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
@@ -270,6 +277,59 @@ public class CachingSchemaStoreTest
assertNotEquals(writer1, writer2);
}
+ /**
+ * A bad/missing Schema Store config at boot leaves {@code publisher} null
(the failure is
+ * logged and swallowed, per {@code reloadPublisherAndPublishSchemas}).
Regression test for
+ * the gap where fixing the {@code configs} table alone never re-triggered
the load — only an
+ * unrelated CQL schema change did. Reacting to {@code
ON_CDC_CONFIGURATION_CHANGED} closes
+ * that gap.
+ */
+ @Test
+ void testOnCdcConfigurationChangedReloadsSchemaStorePublisher() throws
InterruptedException
+ {
+ setupForVersion(CassandraVersion.FOURZERO);
+ when(mockCdcConfig.cdcEnabled()).thenReturn(true);
+
+ AtomicInteger buildPublisherCalls = new AtomicInteger(0);
+ SchemaStorePublisherFactory countingPublisherFactory = kafkaOptions ->
{
+ buildPublisherCalls.incrementAndGet();
+ return mock(TableSchemaPublisher.class);
+ };
+
+ Set<CqlTable> tables = cqlTables(CREATE_STATEMENT);
+ mockCassandraClusterSchemaMonitor = createMockClusterSchema(tables);
+
+ Vertx testVertx = Vertx.vertx();
+ try
+ {
+ cachingSchemaStore = new CachingSchemaStore(testVertx,
mockCassandraClusterSchemaMonitor,
+
spyTableHistoryDatabaseAccessor, mockCdcConfig,
+ mockSidecarCdcStats,
mockSidecarSchema,
+
cqlToAvroSchemaConverter, countingPublisherFactory);
+
+ // ON_SERVER_START registers the ON_CDC_CONFIGURATION_CHANGED
consumer, mirroring real startup.
+ testVertx.eventBus().publish(ON_SERVER_START.address(), "server
started");
+ assertThat(buildPublisherCalls.get()).isEqualTo(0);
+
+ // Registration above is async, so retry the publish until the
consumer picks it up.
+ loopAssert(5, () -> {
+ if (buildPublisherCalls.get() == 0)
+ {
+
testVertx.eventBus().publish(ON_CDC_CONFIGURATION_CHANGED.address(), "Cdc
Configuration Changed");
+ }
+ assertThat(buildPublisherCalls.get()).isEqualTo(1);
+ });
+
+ // Consumer is registered now, so a second config-change event
should trigger another reload.
+
testVertx.eventBus().publish(ON_CDC_CONFIGURATION_CHANGED.address(), "Cdc
Configuration Changed");
+ loopAssert(5, () ->
assertThat(buildPublisherCalls.get()).isEqualTo(2));
+ }
+ finally
+ {
+ testVertx.close();
+ }
+ }
+
public Set<CqlTable> cqlTables(String createStatement)
{
Set<CqlTable> tables =
Set.of(cassandraBridge.buildSchema(createStatement, TEST_KEYSPACE,
REPLICATION_FACTOR, PARTITIONER));
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
