CASSSIDECAR-483: CDC configs in configs table are not wired to SidecarCdcOptions/SidecarStatePersister

CHANGES.txt

@@ -1,5 +1,6 @@
 0.5.0
 -----
+ * Wire CDC configs in configs table to SidecarCdcOptions/SidecarStatePersister (CASSSIDECAR-483)
  * Implement durable operational job tracker (CASSSIDECAR-374)
  * Remove filesystem path from Http response (CASSSIDECAR-477)
  * Add basic configuration retrieval logic to ConfigurationManager (CASSSIDECAR-427)

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

@@ -18,6 +18,7 @@
 
 package org.apache.cassandra.sidecar.cdc;
 
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
@@ -35,6 +36,7 @@
 import org.apache.cassandra.cdc.sidecar.ReplicationFactorSupplier;
 import org.apache.cassandra.cdc.sidecar.SidecarCdc;
 import org.apache.cassandra.cdc.sidecar.SidecarCdcClient;
+import org.apache.cassandra.cdc.sidecar.SidecarCdcOptions;
 import org.apache.cassandra.cdc.sidecar.SidecarCdcStats;
 import org.apache.cassandra.cdc.sidecar.SidecarStatePersister;
 import org.apache.cassandra.cdc.stats.ICdcStats;
@@ -219,10 +221,43 @@ CdcConsumerEntry buildConsumer(@NotNull String jobId,
 
     private @NotNull SidecarStatePersister getSidecarStatePersister()
     {
-        return new SidecarStatePersister(org.apache.cassandra.cdc.sidecar.SidecarCdcOptions.DEFAULT,
+        return new SidecarStatePersister(new ConfigBackedPersisterOptions(conf),
                                          cdcOptions,
                                          SidecarCdcStats.STUB,
                                          cassandraClient,
                                          asyncExecutor);
     }
+
+    /**
+     * Adapts the DB-backed {@link CdcConfig} to the cassandra-analytics-cdc-sidecar
+     * {@link SidecarCdcOptions} interface consumed by {@link SidecarStatePersister}, which only
+     * ever calls {@link SidecarCdcOptions#persistDelay()} on it.
+     *
+     * <p>Previously {@link SidecarStatePersister} was built with {@code SidecarCdcOptions.DEFAULT},
+     * which pinned {@code persistDelay()} to its hardcoded 1000ms interface default regardless of
+     * what operators configured in the "configs" table.
+     *
+     * <p>This is a separate, minimal implementation rather than reusing this class's own
+     * {@code SidecarCdcOptions} (the {@link CdcOptions} implementation consumed by the CDC read
+     * path): the two interfaces have different call sites and no requirement to be backed by the
+     * same object, so keeping them independent avoids one implementation growing overrides it
+     * doesn't need to satisfy the other's contract.
+     */
+    @VisibleForTesting
+    static final class ConfigBackedPersisterOptions implements SidecarCdcOptions
+    {
+        private final CdcConfig conf;
+
+        @VisibleForTesting
+        ConfigBackedPersisterOptions(CdcConfig conf)
+        {
+            this.conf = conf;
+        }
+
+        @Override
+        public Duration persistDelay()
+        {
+            return Duration.ofMillis(conf.persistDelay().toMillis());
+        }
+    }
 }

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

@@ -18,6 +18,7 @@
 
 package org.apache.cassandra.sidecar.cdc;
 
+import java.time.Duration;
 import java.util.Map;
 
 import org.apache.cassandra.bridge.CassandraVersion;
@@ -26,16 +27,22 @@
 import org.apache.cassandra.spark.data.ReplicationFactor;
 
 /**
- * Specific sidecar CDC options
+ * Specific sidecar CDC options, consumed by the CDC read path ({@code SidecarCdc}).
+ *
+ * <p>Delegates the throughput/backpressure-related knobs to {@link CdcConfig} so that they are
+ * live-tunable from the DB-backed "configs" table (via {@link org.apache.cassandra.sidecar.tasks.CdcConfigRefresherNotifierTask})
+ * without a Sidecar restart, instead of silently falling back to the {@link CdcOptions} interface
+ * defaults baked into the cassandra-analytics library.
  */
 public class SidecarCdcOptions implements CdcOptions
 {
-
     private final InstanceMetadataFetcher instanceMetadataFetcher;
+    private final CdcConfig conf;
 
-    public SidecarCdcOptions(InstanceMetadataFetcher instanceMetadataFetcher)
+    public SidecarCdcOptions(InstanceMetadataFetcher instanceMetadataFetcher, CdcConfig conf)
     {
         this.instanceMetadataFetcher = instanceMetadataFetcher;
+        this.conf = conf;
     }
 
 
@@ -59,4 +66,55 @@ public CassandraVersion version()
                 instance -> instance.delegate().nodeSettings().releaseVersion());
         return CassandraVersion.fromVersion(releaseVersion).orElse(CassandraVersion.FOURZERO);
     }
+
+    /**
+     * Add an optional delay between micro-batches, to slow CDC down if it is overwhelming Cassandra
+     * or the downstream Kafka publish stage. Backed by {@code CdcConfig.minDelayBetweenMicroBatches()}
+     * so it can be lowered/raised live via the "configs" table, e.g. to accelerate cdc_raw drain
+     * during a backlog without a restart.
+     */
+    @Override
+    public Duration minimumDelayBetweenMicroBatches()
+    {
+        return Duration.ofMillis(conf.minDelayBetweenMicroBatches().toMillis());
+    }
+
+    /**
+     * Throttles how many commit logs are read per epoch per instance. Backed by
+     * {@code CdcConfig.maxCommitLogsPerInstance()} so it can be raised live to catch up faster
+     * on a backlog, or lowered to bound per-batch memory/duration during burst load.
+     */
+    @Override
+    public int maxCommitLogsPerInstance()
+    {
+        return conf.maxCommitLogsPerInstance();
+    }
+
+    /**
+     * Maximum number of late/un-acked mutation digests held in the CDC watermarker state. Backed
+     * by {@code CdcConfig.maxWatermarkerSize()}.
+     *
+     * <p><b>Caution:</b> {@code CdcState.ReplicaCountSerializer} currently serializes this map's
+     * size with {@code writeShort}/{@code readShort} (signed 16-bit, max 32767). Do not configure
+     * this above 32767 until that serializer is widened to an int, or persisted CDC state can
+     * silently corrupt (observed as a permanent restart-crash-loop in production).
+     */
+    @Override
+    public int maxCdcStateSize()
+    {
+        return conf.maxWatermarkerSize();
+    }
+
+    /**
+     * Maximum age of mutations retained in the CDC watermarker before being purged (and counted via
+     * {@code droppedExpiredMutations}). Backed by {@code CdcConfig.watermarkWindow()} -- previously
+     * this value was entirely unreachable: {@code watermarkWindow()} was read from the DB-backed
+     * config but never consulted by the CDC engine, which instead silently used the 1-hour
+     * {@link CdcOptions#maximumAge()} interface default regardless of what operators configured.
+     */
+    @Override
+    public Duration maximumAge()
+    {
+        return Duration.ofSeconds(conf.watermarkWindow().toSeconds());
+    }
 }

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

@@ -490,9 +490,9 @@ CdcPublisher cdcPublisher(Vertx vertx,
 
     @Provides
     @Singleton
-    public CdcOptions cdcOptions(InstanceMetadataFetcher instanceMetadataFetcher)
+    public CdcOptions cdcOptions(InstanceMetadataFetcher instanceMetadataFetcher, CdcConfig conf)
     {
-        return new SidecarCdcOptions(instanceMetadataFetcher);
+        return new SidecarCdcOptions(instanceMetadataFetcher, conf);
     }
 
     @ProvidesIntoMap

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

@@ -22,12 +22,14 @@
 import java.math.BigInteger;
 import java.net.InetAddress;
 import java.net.UnknownHostException;
+import java.time.Duration;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
 
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -42,6 +44,7 @@
 import org.apache.cassandra.cdc.stats.ICdcStats;
 import org.apache.cassandra.sidecar.cluster.instance.InstanceMetadata;
 import org.apache.cassandra.sidecar.common.server.cluster.locator.TokenRange;
+import org.apache.cassandra.sidecar.common.server.utils.MillisecondBoundConfiguration;
 import org.apache.cassandra.sidecar.concurrent.TaskExecutorPool;
 import org.apache.cassandra.sidecar.coordination.RangeManager;
 import org.apache.cassandra.sidecar.db.CdcDatabaseAccessor;
@@ -369,6 +372,25 @@ void testGetInstanceIdReturnsMinusOneWhenInstanceNotFound()
         assertThat(cdcManager.getInstanceId(unknownIp)).isEqualTo(-1);
     }
 
+    /**
+     * Regression guard: {@code SidecarStatePersister} was previously built with the
+     * cassandra-analytics-cdc-sidecar {@code SidecarCdcOptions.DEFAULT}, which pinned
+     * {@code persistDelay()} to its hardcoded 1000ms interface default regardless of what
+     * operators configured in the "configs" table. {@link CdcManager.ConfigBackedPersisterOptions}
+     * fixes this by delegating {@code persistDelay()} straight to {@link CdcConfig}; this test
+     * uses a value that differs from both the interface default (1000ms) and the
+     * {@code CdcConfigImpl} default (also 1000ms) so a pass proves real delegation.
+     */
+    @Test
+    void configBackedPersisterOptionsDelegatesPersistDelayToCdcConfig()
+    {
+        when(cdcConfig.persistDelay()).thenReturn(new MillisecondBoundConfiguration(2500, TimeUnit.MILLISECONDS));
+
+        CdcManager.ConfigBackedPersisterOptions persisterOptions = new CdcManager.ConfigBackedPersisterOptions(cdcConfig);
+
+        assertThat(persisterOptions.persistDelay()).isEqualTo(Duration.ofMillis(2500));
+    }
+
     // Helper methods
 
     private TokenRange mockTokenRange(BigInteger start, BigInteger end)

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

@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.cdc;
+
+import java.time.Duration;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import org.apache.cassandra.sidecar.common.server.utils.MillisecondBoundConfiguration;
+import org.apache.cassandra.sidecar.common.server.utils.SecondBoundConfiguration;
+import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * Unit tests for {@link SidecarCdcOptions}.
+ *
+ * <p>These specifically guard against the throughput/backpressure knobs silently falling back to
+ * the hardcoded {@code CdcOptions} interface defaults (1000ms micro-batch delay, 4 commit logs
+ * per instance, 200000 max state size, 1 hour max age) instead of the DB-backed {@link CdcConfig}
+ * value an operator configures in the "configs" table. Every value asserted here is deliberately
+ * chosen to differ from both the {@code CdcOptions} interface default and the {@link CdcConfig}
+ * default, so a passing test proves real delegation rather than two defaults coincidentally
+ * matching.
+ */
+class SidecarCdcOptionsTest
+{
+    private CdcConfig conf;
+    private SidecarCdcOptions options;
+
+    @BeforeEach
+    void setUp()
+    {
+        conf = mock(CdcConfig.class);
+        InstanceMetadataFetcher instanceMetadataFetcher = mock(InstanceMetadataFetcher.class);
+        options = new SidecarCdcOptions(instanceMetadataFetcher, conf);
+    }
+
+    @Test
+    void minimumDelayBetweenMicroBatchesDelegatesToCdcConfig()
+    {
+        when(conf.minDelayBetweenMicroBatches()).thenReturn(new MillisecondBoundConfiguration(250, TimeUnit.MILLISECONDS));
+
+        assertThat(options.minimumDelayBetweenMicroBatches()).isEqualTo(Duration.ofMillis(250));
+    }
+
+    @Test
+    void maxCommitLogsPerInstanceDelegatesToCdcConfig()
+    {
+        when(conf.maxCommitLogsPerInstance()).thenReturn(16);
+
+        assertThat(options.maxCommitLogsPerInstance()).isEqualTo(16);
+    }
+
+    @Test
+    void maxCdcStateSizeDelegatesToCdcConfigMaxWatermarkerSize()
+    {
+        when(conf.maxWatermarkerSize()).thenReturn(12345);
+
+        assertThat(options.maxCdcStateSize()).isEqualTo(12345);
+    }
+
+    @Test
+    void maximumAgeDelegatesToCdcConfigWatermarkWindow()
+    {
+        // Regression guard: watermarkWindow() was previously read from the DB-backed config but
+        // never consulted anywhere, so operators configuring it had no actual effect -- the CDC
+        // engine silently used the 1-hour CdcOptions interface default instead.
+        when(conf.watermarkWindow()).thenReturn(new SecondBoundConfiguration(120, TimeUnit.SECONDS));
+
+        assertThat(options.maximumAge()).isEqualTo(Duration.ofSeconds(120));
+    }
+}