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 66c076cbf..58c1bf7b4 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 @@ 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 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 2d48ace1f..c8a15cb53 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 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). + * + *

Wire format note: the map size (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> { public Map 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 replicaCounts = new HashMap<>(numUpdates); for (int i = 0; i < numUpdates; i++) { @@ -413,7 +429,7 @@ public Map read(Kryo kryo, Input in, Cla public void write(Kryo kryo, Output out, Map o) { // write replica counts for late mutations - out.writeShort(o.size()); + out.writeInt(o.size()); for (Map.Entry entry : o.entrySet()) { PartitionUpdateWrapper.Digest digest = entry.getKey();