CASSSIDECAR-243: Implementation of CDCPublisher #294

bbotella · 2025-12-01T22:28:55Z

Patch by James Berragan, Jyothsna Konisa, Bernardo Botella; reviewed by for CASSSIDECAR-243

jyothsnakonisa

Looks good overall, leaving some comments.

server/src/main/java/org/apache/cassandra/sidecar/cdc/CachingSchemaStore.java

server/src/main/java/org/apache/cassandra/sidecar/modules/CdcModule.java

server/src/main/java/org/apache/cassandra/sidecar/tasks/CdcConfigRefresherNotifierTask.java

server/src/main/java/org/apache/cassandra/sidecar/tasks/ClusterTopologyMonitor.java

jyothsnakonisa · 2025-12-10T00:02:19Z

server/src/test/java/org/apache/cassandra/sidecar/cdc/CdcConfigImplTest.java

    }

-    @Test
-    void testConfigChanged() throws Exception


Why did you remove this test? I think this is valuable to have a test for checking callback and config updating when config is changed.

This became a periodic task, and the callback was moved to a vertx signal instead.

But I think the test is sill valid right? may be we should have the test to check if the vertx signal is made when config is changed

jyothsnakonisa · 2025-12-10T23:32:17Z

Thanks @bbotella for addressing my comments. There is this comment #294 (comment) which is not resolved. We need to look into that codepath as it is critical for leader election and mutation publishing.

Also, can you please check if there is a way to add an end to end test that checks if the mutations are published(in-memory) might not need to have kafka for the test.

Other than those two everything else looks good to me.

jyothsnakonisa

Thanks Bernardo for the patch!

jberragan · 2025-12-20T07:35:10Z

server/src/main/java/org/apache/cassandra/sidecar/codecs/BigIntegerCodec.java

+     */
+    public BigInteger decodeFromWire(int pos, Buffer buf)
+    {
+        return decodeFromWire(new MutableInt(pos), buf);


I think the MutableInt is only required if we need to track the change in position in a wrapper serializer, this would be marginally faster as:

Suggested change

return decodeFromWire(new MutableInt(pos), buf);

return new BigInteger(CommonCodecs.BYTE_ARRAY.decodeFromWire(pos, buf));

jberragan · 2025-12-20T09:37:42Z

server/src/main/java/org/apache/cassandra/sidecar/cdc/CdcConfigImpl.java

-    private volatile Map<String, String> kafkaConfigMappings = Map.of();
-    private volatile Map<String, String> cdcConfigMappings = Map.of();
+
+    private Map<String, String> kafkaConfigMappings = Map.of();


These look like they should be volatile

There not accessed with synchronized so better to make them volatile

jberragan · 2025-12-20T09:49:55Z

server/src/main/java/org/apache/cassandra/sidecar/cdc/CdcManager.java

+                                    }
+                                });
+                            }))
+                   .collect(Collectors.toList());


Result of .collect(Collectors.toList()); is ignored/not used.

jberragan · 2025-12-20T09:56:03Z

server/src/main/java/org/apache/cassandra/sidecar/cdc/CdcManager.java

+        }
+
+        // NEW: Deduplicate by (instanceId, tokenRange) to prevent duplicate consumers
+        Map<String, SidecarCdc> uniqueConsumers = new HashMap<>();


Suggested change

Map<String, SidecarCdc> uniqueConsumers = new HashMap<>();

Map<String, SidecarCdc> uniqueConsumers = new HashMap<>(ownedRanges.values().stream().mapToInt(Set::size).sum());

jberragan · 2025-12-20T10:01:55Z

server/src/main/java/org/apache/cassandra/sidecar/cdc/CdcManager.java

+        return -1;
+    }
+
+    private boolean resolveToSameAddress(String host1, String host2)


It seems like this only expects ip addresses to be input, and so will not perform DNS resolution. Can we make that explicit? We can also make the method static and add some unit tests here.

jberragan · 2025-12-20T10:02:27Z

server/src/main/java/org/apache/cassandra/sidecar/cdc/CdcManager.java

+    {
+        for (InstanceMetadata instance : instanceFetcher.allLocalInstances())
+        {
+            String configuredHost = instance.ipAddress();


This is an ip address, no?

Suggested change

String configuredHost = instance.ipAddress();

String configuredIpAddress = instance.ipAddress();

jberragan · 2025-12-21T08:08:51Z

server/src/main/java/org/apache/cassandra/sidecar/cdc/CdcPublisher.java

+    private final ICdcStats cdcStats;
+    private final SidecarConfiguration sidecarConfiguration;
+    private CdcManager cdcManager;
+    private Serializer<CdcEvent> avroSerializer;


Can be final:

Suggested change

private Serializer<CdcEvent> avroSerializer;

private final Serializer<CdcEvent> avroSerializer;

private final Provider<RangeManager> rangeManagerProvider;

jberragan

Mostly nits, +1 (nb)

yifan-c

When CDC is disabled,

yifan-c · 2026-01-08T23:54:44Z

server/src/main/java/org/apache/cassandra/sidecar/cdc/CachingSchemaStore.java

+            loadPublisher();
+            publishSchemas();


Those 2 methods are called in every iteration. Looks like they are included in the for-loop by mistake.

yifan-c · 2026-01-08T23:58:50Z

server/src/main/java/org/apache/cassandra/sidecar/cdc/CachingSchemaStore.java

+    private void loadPublisher()
+    {
+        KafkaOptions kafkaOptions = () -> cdcConfig.kafkaConfigs();
+        this.publisher = SchemaStorePublisherFactory.DEFAULT.buildPublisher(kafkaOptions);


publisher is a Closeable object. We should close the previous object when assigning a new one.

yifan-c · 2026-01-08T23:59:25Z

server/src/main/java/org/apache/cassandra/sidecar/cdc/CdcManager.java

+                                    }
+                                });
+                            }))
+                   .collect(Collectors.toList());;


yifan-c · 2026-01-09T00:04:29Z

server/src/main/java/org/apache/cassandra/sidecar/server/SidecarServerEvents.java

+    ON_CDC_CONFIGURATION_CHANGED,
+    ON_CDC_CACHE_WARMED_UP,
+    ON_CDC_CONFIG_MAPPINGS_CHANGED,


Add javadoc

yifan-c · 2026-01-09T00:04:45Z

server/src/main/java/org/apache/cassandra/sidecar/modules/CdcModule.java

+                                    KafkaConfigAccessor kafkaConfigAccessor,
+                                    CdcConfigAccessor cdcConfigAccessor)


indentation

yifan-c · 2026-01-09T00:10:08Z

server/src/main/java/org/apache/cassandra/sidecar/cdc/CachingSchemaStore.java

+                       SidecarSchema sidecarSchema,
+                       CqlToAvroSchemaConverter cqlToAvroSchemaConverter)
+    {
+        super();


super() is redundant, because it is just implementing the interface SchemaStore

yifan-c · 2026-01-09T00:12:57Z

server/src/main/java/org/apache/cassandra/sidecar/cdc/CachingSchemaStore.java

+        this.avroSchemasCache.putAll(createSchemaCache(cassandraClusterSchemaMonitor.getCdcTables()));
+        AvroSchemas.registerLogicalTypes();
+        cassandraClusterSchemaMonitor.addSchemaChangeListener(this::onSchemaChanged);
+        this.vertx = vertx;
+        this.cdcConfig = cdcConfig;
+        this.sidecarCdcStats = sidecarCdcStats;
+
+        configureSidecarServerEventListeners();


When CDC is disabled, we should skip those all together. It does not make sense to register listeners, etc.

Ideally, we should not mount the CdcModule entirely, if the feature is not enabled. It is outside of the scope of this PR.

I will create a Jira

Thanks. Just to clarify that conditionally register the listeners are still wanted in this patch.

yifan-c · 2026-01-09T00:14:01Z

server/src/main/java/org/apache/cassandra/sidecar/cdc/CdcPublisher.java

+        vertx.eventBus().localConsumer(RangeManager.RangeManagerEvents.ON_TOKEN_RANGE_CHANGED.address(), this);
+        vertx.eventBus().localConsumer(RangeManager.LeadershipEvents.ON_TOKEN_RANGE_GAINED.address(), this);
+        vertx.eventBus().localConsumer(RangeManager.LeadershipEvents.ON_TOKEN_RANGE_LOST.address(), this);
+        vertx.eventBus().localConsumer(ON_SERVER_STOP.address(), this);
+        vertx.eventBus().localConsumer(ON_CDC_CACHE_WARMED_UP.address(), this);
+        vertx.eventBus().localConsumer(ON_CDC_CONFIGURATION_CHANGED.address(), new ConfigChangedHandler());


Skip those when CDC is disabled.

Should this be part of the Jira mentioned in previous comment?

Those should be behind the condition whether cdc is enabled.

yifan-c · 2026-01-09T00:25:17Z

server/src/main/java/org/apache/cassandra/sidecar/cdc/CdcPublisher.java

+        KafkaProducer<String, byte[]> producer = new KafkaProducer<>(conf.kafkaConfigs());
+        KafkaPublisher kafkaPublisher = new KafkaPublisher(TopicSupplier.staticTopicSupplier(conf.kafkaTopic()),


Both producer and kafkaPublisher are Closeable. I do not see them get closed anywhere. We are create a new eventConsumer on every publisher restart, meaning leaks.

CASSSIDECAR-242: Implementation of CachingSchemaStore House cleanup WIP commit Working CDC on Sidecar Latest changes Working CDC Working Working

bbotella requested a review from jyothsnakonisa December 1, 2025 22:28

jyothsnakonisa reviewed Dec 10, 2025

View reviewed changes

jyothsnakonisa approved these changes Dec 19, 2025

View reviewed changes

bbotella force-pushed the CASSSIDECAR-243-CDCPublisher branch from 73efd40 to b594d40 Compare December 19, 2025 18:22

jberragan reviewed Dec 20, 2025

View reviewed changes

jberragan reviewed Dec 21, 2025

View reviewed changes

jberragan approved these changes Dec 21, 2025

View reviewed changes

bbotella force-pushed the CASSSIDECAR-243-CDCPublisher branch from 87d98d5 to 363a547 Compare January 7, 2026 16:37

yifan-c requested changes Jan 9, 2026

View reviewed changes

yifan-c approved these changes Jan 9, 2026

View reviewed changes

bbotella added 14 commits January 9, 2026 22:08

CASSSIDECAR-242: Implementation of CachingSchemaStore

e9c0b08

CASSSIDECAR-242: Implementation of CachingSchemaStore House cleanup WIP commit Working CDC on Sidecar Latest changes Working CDC Working Working

Revert not needed changes

dc9ef3b

Fix RAT

ba0641a

Fix checkstyle

b1214c1

Fix tests

670619a

Add readme info

b94ccc2

Applies feedback

7c1e81c

Add missing license

6865f60

Fix deferred injection

1c8f4b0

Re add config changed test

02bc399

Fix checkstyle

cb4f581

Fix checkstyle

bbcc22e

Add Changes.txt

d60ed0d

Adds feedback

881f55c

bbotella added 6 commits January 9, 2026 22:09

Fix tests

3b5fc76

Revert sidecar.yaml changes

ca6faee

Remove not needed todo comment

296853d

Close objects

5ab5732

Skip event registration when CDC is not enabled

17825d2

Avoid final event registration if CDC is not enabled

8c25504

bbotella force-pushed the CASSSIDECAR-243-CDCPublisher branch from 5825499 to 8c25504 Compare January 9, 2026 21:09

bbotella merged commit 1ee13fb into apache:trunk Jan 9, 2026
29 of 30 checks passed

	return decodeFromWire(new MutableInt(pos), buf);
	return new BigInteger(CommonCodecs.BYTE_ARRAY.decodeFromWire(pos, buf));

	Map<String, SidecarCdc> uniqueConsumers = new HashMap<>();
	Map<String, SidecarCdc> uniqueConsumers = new HashMap<>(ownedRanges.values().stream().mapToInt(Set::size).sum());

	String configuredHost = instance.ipAddress();
	String configuredIpAddress = instance.ipAddress();

	private Serializer<CdcEvent> avroSerializer;
	private final Serializer<CdcEvent> avroSerializer;
	private final Provider<RangeManager> rangeManagerProvider;

		KafkaConfigAccessor kafkaConfigAccessor,
		CdcConfigAccessor cdcConfigAccessor)

		KafkaProducer<String, byte[]> producer = new KafkaProducer<>(conf.kafkaConfigs());
		KafkaPublisher kafkaPublisher = new KafkaPublisher(TopicSupplier.staticTopicSupplier(conf.kafkaTopic()),

CASSSIDECAR-243: Implementation of CDCPublisher #294

CASSSIDECAR-243: Implementation of CDCPublisher #294

Uh oh!

Conversation

bbotella commented Dec 1, 2025

Uh oh!

jyothsnakonisa left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jyothsnakonisa commented Dec 10, 2025

Uh oh!

jyothsnakonisa left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jberragan left a comment

Choose a reason for hiding this comment

Uh oh!

yifan-c left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees