CASSANALYTICS-184 : CdcState.ReplicaCountSerializer map-size overflow corrupts persisted CDC state

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 @@ private static void testCdcStateSerialization(CdcState expected)
         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
+        };
+    }
 }

cassandra-analytics-common/src/main/java/org/apache/cassandra/cdc/state/CdcState.java

@@ -395,12 +395,28 @@ public static CdcState deserialize(Kryo kryo, CompressionUtil compressionUtil, b
         }
     }
 
+    /**
+     * 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 Map<PartitionUpdateWrapper.Digest, Integer> read(Kryo kryo, Input in, Cla
         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();