You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by jm...@apache.org on 2026/07/29 20:11:47 UTC
(cassandra-ecosystem) 01/01: CASSANALYTICS-184 : CdcState.ReplicaCountSerializer map-size overflow corrupts persisted CDC state (#225)
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 <jk...@apache.org>
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: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org