This is an automated email from the ASF dual-hosted git repository. jmckenzie-dev pushed a commit to branch analytics/trunk in repository https://gitbox.apache.org/repos/asf/cassandra-ecosystem.git
commit c0aab1d44a510fd26527779747d22e8b378d00e6 Author: Jyothsna konisa <[email protected]> AuthorDate: Tue Jul 21 07:03:59 2026 -0700 CASSANALYTICS-184 : CdcState.ReplicaCountSerializer map-size overflow corrupts persisted CDC state (#225) Patch by Jyothsna Konisa; Reviewed by Josh McKenzie for CASSANALYTICS-184 --- CHANGES.txt | 1 + .../cdc/model/CdcKryoSerializationTests.java | 46 ++++++++++++++++++++++ .../org/apache/cassandra/cdc/state/CdcState.java | 22 +++++++++-- 3 files changed, 66 insertions(+), 3 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 49ed57ae..811417a7 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,5 +1,6 @@ 0.5.0 ----- + * CdcState.ReplicaCountSerializer map-size overflow corrupts persisted CDC state (CASSANALYTICS-184) * SSTable-version-based bridge determination (CASSANALYTICS-24) * Upgrade sidecar version to 0.4.0 (CASSANALYTICS-176) * Exclude IP address from RingInstance equality so node replacement does not fail bulk write jobs (CASSANALYTICS-175) diff --git a/cassandra-analytics-cdc/src/test/java/org/apache/cassandra/cdc/model/CdcKryoSerializationTests.java b/cassandra-analytics-cdc/src/test/java/org/apache/cassandra/cdc/model/CdcKryoSerializationTests.java index 66c076cb..58c1bf7b 100644 --- a/cassandra-analytics-cdc/src/test/java/org/apache/cassandra/cdc/model/CdcKryoSerializationTests.java +++ b/cassandra-analytics-cdc/src/test/java/org/apache/cassandra/cdc/model/CdcKryoSerializationTests.java @@ -20,6 +20,8 @@ package org.apache.cassandra.cdc.model; import java.math.BigInteger; +import java.util.HashMap; +import java.util.Map; import com.google.common.collect.ImmutableMap; import org.junit.jupiter.api.Test; @@ -110,4 +112,48 @@ public class CdcKryoSerializationTests assertThat(deserialized).isNotNull(); assertThat(deserialized).isEqualTo(expected); } + + /** + * Regression test for the {@code ReplicaCountSerializer} short-overflow bug: the replica-count + * map size used to be written/read as a signed short (max 32767), so a watermarker larger than + * that (reachable via {@code CdcOptions.maxCdcStateSize()}/{@code maxWatermarkerSize()}, both + * configurable well above 32767) would silently overflow on write and deserialize with a + * corrupted (potentially negative) size, observed in production as a permanent + * restart-crash-loop ({@code IllegalArgumentException: Illegal initial capacity}). The map size + * is now a signed int, so this must round-trip correctly for a map with more than 32767 entries. + */ + @Test + public void testCdcStateReplicaCountMapLargerThanShortOverflows() + { + int size = Short.MAX_VALUE + 1000; // comfortably past the old 32767 short limit + Map<PartitionUpdateWrapper.Digest, Integer> replicaCount = new HashMap<>(size); + for (int i = 0; i < size; i++) + { + PartitionUpdateWrapper.Digest digest = new PartitionUpdateWrapper.Digest("ks1", + "tb1", + TimeUtils.nowMicros(), + intToBytes(i), + 500, + BigInteger.ONE); + replicaCount.put(digest, i % 128); // replica count is serialized as a single byte + } + + CdcState expected = CdcState.of(1L, + TokenRange.openClosed(BigInteger.ONE, BigInteger.TEN), + CommitLogMarkers.EMPTY, + replicaCount); + assertThat(expected.size()).isEqualTo(size); + + testCdcStateSerialization(expected); + } + + private static byte[] intToBytes(int value) + { + return new byte[]{ + (byte) (value >>> 24), + (byte) (value >>> 16), + (byte) (value >>> 8), + (byte) value + }; + } } diff --git a/cassandra-analytics-common/src/main/java/org/apache/cassandra/cdc/state/CdcState.java b/cassandra-analytics-common/src/main/java/org/apache/cassandra/cdc/state/CdcState.java index 2d48ace1..c8a15cb5 100644 --- a/cassandra-analytics-common/src/main/java/org/apache/cassandra/cdc/state/CdcState.java +++ b/cassandra-analytics-common/src/main/java/org/apache/cassandra/cdc/state/CdcState.java @@ -395,12 +395,28 @@ public class CdcState } } + /** + * Serializes the replica-count map for late/un-acked mutations awaiting sufficient replicas. + * Map key = mutation digest; value = replica count seen so far for that mutation (always small, + * bounded by replication factor, so kept as a single byte -- unrelated to the bug below). + * + * <p><b>Wire format note:</b> the map <em>size</em> (entry count, i.e. number of distinct + * mutations tracked) is written as a signed int -- previously a signed short (max 32767 + * entries), which silently overflowed and corrupted persisted state once + * {@code CdcOptions.maxCdcStateSize()}/{@code maxWatermarkerSize()} configured a larger value + * (both default well above 32767), observed in production as a permanent restart-crash-loop. + * {@code int} is sufficient since the configured max size is always far below + * {@code Integer.MAX_VALUE}; {@code long} would be unwarranted here. This is a wire-format + * change: state persisted by the old (short) format cannot be read by this version and vice + * versa. Rolling upgrades should account for this (e.g. purge/rebuild persisted CDC state + * rather than carry it across the upgrade boundary). + */ public static class ReplicaCountSerializer extends com.esotericsoftware.kryo.Serializer<Map<PartitionUpdateWrapper.Digest, Integer>> { public Map<PartitionUpdateWrapper.Digest, Integer> read(Kryo kryo, Input in, Class type) { - // read replica counts - int numUpdates = in.readShort(); + // numUpdates = number of distinct mutation digests tracked (map size), not a replica count + int numUpdates = in.readInt(); Map<PartitionUpdateWrapper.Digest, Integer> replicaCounts = new HashMap<>(numUpdates); for (int i = 0; i < numUpdates; i++) { @@ -413,7 +429,7 @@ public class CdcState public void write(Kryo kryo, Output out, Map<PartitionUpdateWrapper.Digest, Integer> o) { // write replica counts for late mutations - out.writeShort(o.size()); + out.writeInt(o.size()); for (Map.Entry<PartitionUpdateWrapper.Digest, Integer> entry : o.entrySet()) { PartitionUpdateWrapper.Digest digest = entry.getKey(); --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
