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 f8fee7b91 CASSSIDECAR-492: CDC log_only mode should not require Kafka
producer … (#389)
f8fee7b91 is described below
commit f8fee7b91c3e64725d467124b13de88fbb3afe9f
Author: Shruti Sekaran <[email protected]>
AuthorDate: Fri Oct 9 18:30:43 2026 -0700
CASSSIDECAR-492: CDC log_only mode should not require Kafka producer …
(#389)
Patch by Shruti Sekaran; reviewed by Jyothsna Konisa, Bernardo Botella,
Yifan Cai for CASSSIDECAR-492
---
CHANGES.txt | 1 +
README.md | 3 ++
.../cassandra/sidecar/cdc/CdcConfigImpl.java | 4 +--
.../apache/cassandra/sidecar/cdc/CdcPublisher.java | 8 ++++++
.../cassandra/sidecar/cdc/CdcConfigImplTest.java | 32 ++++++++++++++++++++++
.../cassandra/sidecar/cdc/CdcPublisherTests.java | 29 +++++++++++++++++++-
6 files changed, 73 insertions(+), 4 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index d1df53b35..28fb6f030 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,5 +1,6 @@
0.5.0
-----
+ * CDC log_only mode should not require Kafka producer configuration
(CASSSIDECAR-492)
* Report unknown node status as "Unknown" instead of "?" in ring and token
range replica responses (CASSSIDECAR-506)
* Remove unused Instance Metadata Fetcher from CdcDatabaseAccessor
(CASSSIDECAR-497)
* Fix IPV6 regex to recognize bracketed addresses in gossip info host headers
(CASSSIDECAR-503)
diff --git a/README.md b/README.md
index 3f4a7931d..728fc8c45 100644
--- a/README.md
+++ b/README.md
@@ -209,6 +209,9 @@ curl --request PUT \
- `TABLE`: Format as `{topic}-{table}`
- `MAP`: Use custom topic mapping (advanced)
- `topic`: Base Kafka topic name for CDC events.
+- `log_only`: When `true`, CDC mutations are logged instead of published to
Kafka, and no Kafka
+ producer configuration is required. Mutations processed in this mode are not
published to Kafka
+ later, even after `log_only` is turned off.
#### 5. Configure Kafka Producer
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/cdc/CdcConfigImpl.java
b/server/src/main/java/org/apache/cassandra/sidecar/cdc/CdcConfigImpl.java
index 3e493aaa0..9f37a4dd6 100644
--- a/server/src/main/java/org/apache/cassandra/sidecar/cdc/CdcConfigImpl.java
+++ b/server/src/main/java/org/apache/cassandra/sidecar/cdc/CdcConfigImpl.java
@@ -94,10 +94,8 @@ public class CdcConfigImpl implements CdcConfig
@Override
public boolean isConfigReady()
{
- kafkaConfigs();
- cdcConfigs();
return cdcConfigAccessor.isAvailable()
- && !kafkaConfigMappings.isEmpty()
+ && (logOnly() || !kafkaConfigMappings.isEmpty())
&& !cdcConfigMappings.isEmpty();
}
diff --git
a/server/src/main/java/org/apache/cassandra/sidecar/cdc/CdcPublisher.java
b/server/src/main/java/org/apache/cassandra/sidecar/cdc/CdcPublisher.java
index 5a1105405..f52cdf1fa 100644
--- a/server/src/main/java/org/apache/cassandra/sidecar/cdc/CdcPublisher.java
+++ b/server/src/main/java/org/apache/cassandra/sidecar/cdc/CdcPublisher.java
@@ -141,6 +141,14 @@ public class CdcPublisher implements
Handler<Message<Object>>, PeriodicTask
instanceMetadataFetcher.callOnFirstAvailableInstance(instance ->
instance.delegate().nodeSettings()).releaseVersion()
).getVersion();
+
+ if (conf.logOnly())
+ {
+ LOGGER.info("CDC log_only is set; mutations will be logged and not
published to Kafka");
+ CdcLogMode.init(key ->
TypeCache.get(version).getType(key.keyspace, key.type));
+ return event -> CdcLogMode.FULL.info(LOGGER, "CDC log_only",
event, null);
+ }
+
this.kafkaPublisher = KafkaPublisher.create(version,
buildTopicSupplier(conf),
conf.kafkaConfigs(),
diff --git
a/server/src/test/java/org/apache/cassandra/sidecar/cdc/CdcConfigImplTest.java
b/server/src/test/java/org/apache/cassandra/sidecar/cdc/CdcConfigImplTest.java
index c53cd08b5..379030842 100644
---
a/server/src/test/java/org/apache/cassandra/sidecar/cdc/CdcConfigImplTest.java
+++
b/server/src/test/java/org/apache/cassandra/sidecar/cdc/CdcConfigImplTest.java
@@ -141,6 +141,38 @@ class CdcConfigImplTest
assertThat(cdcConfig.persistDelay()).isEqualTo(new
MillisecondBoundConfiguration(5, TimeUnit.SECONDS));
}
+ /**
+ * With {@code log_only} set there is no Kafka producer configuration to
load, so readiness must
+ * not require it. Note that {@link CdcConfigImpl#isConfigReady()} has no
production caller — it
+ * is not an enforced gate, and this pins the policy rather than any
behaviour.
+ */
+ @Test
+ void testIsConfigReadyWithoutKafkaConfigsWhenLogOnly() throws
InterruptedException
+ {
+ // Dedicated Vertx instance so the codec registration does not depend
on test ordering
+ Vertx testVertx = Vertx.vertx();
+ try
+ {
+ CdcConfigAccessor cdcConfigAccessor = mockCdcConfigAccessor();
+ CdcConfigRefresherNotifierTask.ConfigMappings configMappings = new
CdcConfigRefresherNotifierTask.ConfigMappings();
+ // Kafka mappings deliberately left empty, mirroring a broker-less
deployment
+ configMappings.setCdcConfigMappings(Map.of("log_only", "true",
+ "datacenter", "DC1"));
+
+ CdcConfigImpl cdcConfig = new CdcConfigImpl(testVertx,
cdcConfigAccessor);
+
testVertx.eventBus().registerDefaultCodec(CdcConfigRefresherNotifierTask.ConfigMappings.class,
CdcConfigMappingsCodec.INSTANCE);
+
testVertx.eventBus().publish(ON_CDC_CONFIG_MAPPINGS_CHANGED.address(),
configMappings);
+
+ loopAssert(5, () -> assertThat(cdcConfig.logOnly()).isTrue());
+ assertThat(cdcConfig.kafkaConfigs()).isEmpty();
+ assertThat(cdcConfig.isConfigReady()).isTrue();
+ }
+ finally
+ {
+ testVertx.close();
+ }
+ }
+
@Test
void testConfigChanged() throws InterruptedException
{
diff --git
a/server/src/test/java/org/apache/cassandra/sidecar/cdc/CdcPublisherTests.java
b/server/src/test/java/org/apache/cassandra/sidecar/cdc/CdcPublisherTests.java
index 8bf52619e..bc252960c 100644
---
a/server/src/test/java/org/apache/cassandra/sidecar/cdc/CdcPublisherTests.java
+++
b/server/src/test/java/org/apache/cassandra/sidecar/cdc/CdcPublisherTests.java
@@ -70,6 +70,7 @@ import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.when;
/**
@@ -159,6 +160,22 @@ public class CdcPublisherTests
assertThat(result).isInstanceOf(CdcEventConsumer.class);
}
+ /**
+ * Log-only must not require Kafka producer configuration, so the producer
factory must never be
+ * reached. Asserting only the returned type would still pass if a
producer were built and then
+ * discarded.
+ */
+ @Test
+ void testEventConsumerBuildsNoKafkaProducerWhenLogOnly()
+ {
+ setupVersionMocks();
+ when(cdcConfig.logOnly()).thenReturn(true);
+
+ cdcPublisher.eventConsumer(cdcConfig);
+
+ verifyNoInteractions(kafkaProducerFactory);
+ }
+
/**
* Verifies that construction captures the Cdc enabled/disabled lifecycle
stat depending on
* {@link CdcConfig#cdcEnabled()} at the time the {@link CdcPublisher} is
built, mirroring the
@@ -245,6 +262,17 @@ public class CdcPublisherTests
when(cdcConfig.failOnRecordTooLargeError()).thenReturn(false);
when(cdcConfig.failOnKafkaError()).thenReturn(true);
+ setupVersionMocks();
+
when(kafkaProducerFactory.create(any())).thenReturn(mock(KafkaProducer.class));
+ }
+
+ /**
+ * Stubs only the chain {@link CdcPublisher#eventConsumer} walks to
resolve the Cassandra
+ * version. Kept separate from {@link #setupMocks} so that the log-only
case can assert the
+ * Kafka producer factory was never touched, without a stubbing on it to
explain away.
+ */
+ private void setupVersionMocks()
+ {
InstanceMetadata mockInstance = mock(InstanceMetadata.class,
RETURNS_DEEP_STUBS);
when(mockInstance.delegate().nodeSettings().releaseVersion()).thenReturn("4.1.0");
doAnswer(invocation -> {
@@ -255,7 +283,6 @@ public class CdcPublisherTests
CassandraBridge mockBridge = mock(CassandraBridge.class);
when(mockBridge.getVersion()).thenReturn(CassandraVersion.FOURONE);
when(cassandraBridgeFactory.get(anyString())).thenReturn(mockBridge);
-
when(kafkaProducerFactory.create(any())).thenReturn(mock(KafkaProducer.class));
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
