This is an automated email from the ASF dual-hosted git repository.
bbotella pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra-analytics.git
The following commit(s) were added to refs/heads/trunk by this push:
new 769f4bc5 CASSANALYTICS-196: Skip CDC-irrelevant mutations on
deserialization failure (#241)
769f4bc5 is described below
commit 769f4bc525cb47bffd0322f164b0ff3abdccbd5a
Author: Yifan Cai <[email protected]>
AuthorDate: Wed Oct 7 01:57:06 2026 -0700
CASSANALYTICS-196: Skip CDC-irrelevant mutations on deserialization failure
(#241)
Patch by Yifan Cai; Reviewed by Bernardo Botella, Jyothsna Konisa for
CASSANALYTICS-196
* address review comments
---
CHANGES.txt | 1 +
.../org/apache/cassandra/cdc/stats/ICdcStats.java | 10 ++
.../db/commitlog/BufferingCommitLogReader.java | 53 +++++++++
.../BufferingCommitLogReaderTableContextTests.java | 129 +++++++++++++++++++++
4 files changed, 193 insertions(+)
diff --git a/CHANGES.txt b/CHANGES.txt
index 7f691430..092926ae 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,5 +1,6 @@
0.5.0
-----
+ * Skip CDC-irrelevant mutations on deserialization failure (CASSANALYTICS-196)
* Opening one keyspace instance rebuilds every keyspace instance in the
Cassandra 6.0 bridge (CASSANALYTICS-202)
* CDC logs NPE for deleted column values (CASSANALYTICS-178)
* Add Cassandra 6.0 support (CASSANALYTICS-37)
diff --git
a/cassandra-analytics-common/src/main/java/org/apache/cassandra/cdc/stats/ICdcStats.java
b/cassandra-analytics-common/src/main/java/org/apache/cassandra/cdc/stats/ICdcStats.java
index 3098ce4c..2e26bff6 100644
---
a/cassandra-analytics-common/src/main/java/org/apache/cassandra/cdc/stats/ICdcStats.java
+++
b/cassandra-analytics-common/src/main/java/org/apache/cassandra/cdc/stats/ICdcStats.java
@@ -179,6 +179,16 @@ public interface ICdcStats extends
BufferingInputStreamStats<CommitLog>
{
}
+ /**
+ * Called when deserialization of a mutation fails, but the mutation can
be safely ignored because it
+ * doesn't involve a CDC-enabled table
+ *
+ * @param incrCount delta value to add to the count
+ */
+ default void mutationsDeserializeFailedNonCdcCount(long incrCount)
+ {
+ }
+
/**
* Called when a mutation's checksum calculation fails or doesn't match
with expected checksum
*
diff --git
a/cassandra-four-zero-bridge/src/main/java/org/apache/cassandra/db/commitlog/BufferingCommitLogReader.java
b/cassandra-four-zero-bridge/src/main/java/org/apache/cassandra/db/commitlog/BufferingCommitLogReader.java
index 41ac5c5f..9d881b1a 100644
---
a/cassandra-four-zero-bridge/src/main/java/org/apache/cassandra/db/commitlog/BufferingCommitLogReader.java
+++
b/cassandra-four-zero-bridge/src/main/java/org/apache/cassandra/db/commitlog/BufferingCommitLogReader.java
@@ -51,6 +51,9 @@ import org.apache.cassandra.io.util.DataInputBuffer;
import org.apache.cassandra.io.util.FileDataInput;
import org.apache.cassandra.io.util.RandomAccessReader;
import org.apache.cassandra.io.util.RebufferingInputStream;
+import org.apache.cassandra.schema.Schema;
+import org.apache.cassandra.schema.TableId;
+import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.spark.exceptions.TransportFailureException;
import org.apache.cassandra.spark.utils.AsyncExecutor;
import org.apache.cassandra.spark.utils.LoggerHelper;
@@ -568,7 +571,18 @@ public class BufferingCommitLogReader implements
CommitLogReadHandler,
return;
}
+ // TODO: This could crash the process (e.g. Sidecar) to crash. We
may want to handle it gracefully depending on the runtime environment.
JVMStabilityInspector.inspectThrowable(t);
+
+ if (!failedMutationMayInvolveCdcTable(inputBuffer, size))
+ {
+ // No CDC-enabled table is involved – safe to skip and move on
+ logger.trace("Ignoring mutation that failed to deserialize; no
CDC-enabled table involved", "error", t);
+ stats.mutationsDeserializeFailedNonCdcCount(1);
+
+ return;
+ }
+
Path p = Files.createTempFile("mutation", "dat");
try (DataOutputStream out = new
DataOutputStream(Files.newOutputStream(p)))
@@ -641,6 +655,45 @@ public class BufferingCommitLogReader implements
CommitLogReadHandler,
return null;
}
+ /**
+ * Best-effort check for whether a mutation that failed to deserialize may
involve a CDC-enabled table.
+ * When this returns false, the mutation is guaranteed not to involve any
CDC-enabled table, so the failure can
+ * be safely ignored.
+ * Only reads the first {@link TableId} to make the decision, as mutations
contain single partition update in most cases.
+ * If a mutation has more than one partition update, it conservatively
assumes a CDC-enabled table is involved.
+ *
+ * @param inputBuffer raw byte array w/Mutation data
+ * @param size serialized size of the mutation
+ * @return true if the first table touched by the mutation is confirmed
CDC-enabled, or the mutation has more
+ * than one {@link PartitionUpdate} (so the rest can't be safely
checked); false otherwise, including
+ * when the table's metadata can't be found, or even the first
{@link TableId} can't be read
+ */
+ @VisibleForTesting
+ static boolean failedMutationMayInvolveCdcTable(byte[] inputBuffer, int
size)
+ {
+ try (DataInputBuffer in = new DataInputBuffer(inputBuffer, 0, size))
+ {
+ int updateCount = (int) in.readUnsignedVInt();
+ if (updateCount > 1)
+ {
+ // Can't safely reach the remaining updates' TableIds without
deserializing this update's body,
+ // conservatively assume it may involve a CDC-enabled table.
+ return true;
+ }
+ TableId tableId = TableId.deserialize(in);
+ TableMetadata metadata = Schema.instance.getTableMetadata(tableId);
+ if (metadata != null && metadata.params.cdc)
+ {
+ return true;
+ }
+ }
+ catch (Throwable t)
+ {
+ // unable to read the first TableId
+ }
+ return false;
+ }
+
public void close()
{
if (updates == null)
diff --git
a/cassandra-four-zero-bridge/src/test/java/org/apache/cassandra/db/commitlog/BufferingCommitLogReaderTableContextTests.java
b/cassandra-four-zero-bridge/src/test/java/org/apache/cassandra/db/commitlog/BufferingCommitLogReaderTableContextTests.java
new file mode 100644
index 00000000..6ffbdb2a
--- /dev/null
+++
b/cassandra-four-zero-bridge/src/test/java/org/apache/cassandra/db/commitlog/BufferingCommitLogReaderTableContextTests.java
@@ -0,0 +1,129 @@
+/*
+ * 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.db.commitlog;
+
+import java.util.UUID;
+
+import org.junit.jupiter.api.Test;
+
+import org.apache.cassandra.bridge.CassandraBridgeImplementation;
+import org.apache.cassandra.cql3.CQLFragmentParser;
+import org.apache.cassandra.cql3.CqlParser;
+import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.io.util.DataOutputBuffer;
+import org.apache.cassandra.schema.KeyspaceMetadata;
+import org.apache.cassandra.schema.KeyspaceParams;
+import org.apache.cassandra.schema.Schema;
+import org.apache.cassandra.schema.TableId;
+import org.apache.cassandra.schema.TableMetadata;
+import org.apache.cassandra.schema.Types;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class BufferingCommitLogReaderTableContextTests
+{
+ @Test
+ public void
testFailedMutationMayInvolveCdcTableWhenNonCdcTableIdReadableButRestCorrupted()
throws Exception
+ {
+ TableId tableId = registerTable("ks_non_cdc", "tbl", false);
+ byte[] corruptedMutation = buildCorruptedMutation(1, tableId);
+
+ boolean mayInvolveCdcTable =
BufferingCommitLogReader.failedMutationMayInvolveCdcTable(corruptedMutation,
+
corruptedMutation.length);
+
+ assertThat(mayInvolveCdcTable).isFalse();
+ }
+
+ @Test
+ public void
testFailedMutationMayInvolveCdcTableWhenCdcTableIdReadableButRestCorrupted()
throws Exception
+ {
+ TableId tableId = registerTable("ks_cdc", "tbl", true);
+ byte[] corruptedMutation = buildCorruptedMutation(1, tableId);
+
+ boolean mayInvolveCdcTable =
BufferingCommitLogReader.failedMutationMayInvolveCdcTable(corruptedMutation,
+
corruptedMutation.length);
+
+ assertThat(mayInvolveCdcTable).isTrue();
+ }
+
+ @Test
+ public void testFailedMutationMayInvolveCdcTableWhenTableIdUnresolvable()
throws Exception
+ {
+ CassandraBridgeImplementation.setup();
+ TableId unknownTableId = TableId.fromUUID(UUID.randomUUID());
+ byte[] corruptedMutation = buildCorruptedMutation(1, unknownTableId);
+
+ boolean mayInvolveCdcTable =
BufferingCommitLogReader.failedMutationMayInvolveCdcTable(corruptedMutation,
+
corruptedMutation.length);
+
+ // an unresolvable table is treated as not CDC-enabled
+ assertThat(mayInvolveCdcTable).isFalse();
+ }
+
+ @Test
+ public void
testFailedMutationMayInvolveCdcTableWhenMultipleUpdatesEvenIfFirstIsNonCdc()
throws Exception
+ {
+ TableId tableId = registerTable("ks_non_cdc_multi", "tbl", false);
+ byte[] corruptedMutation = buildCorruptedMutation(2, tableId);
+
+ boolean mayInvolveCdcTable =
BufferingCommitLogReader.failedMutationMayInvolveCdcTable(corruptedMutation,
+
corruptedMutation.length);
+
+ // can't safely check the remaining update(s) without deserializing
this one's body, so this is
+ // conservatively treated as possibly involving a CDC-enabled table
+ assertThat(mayInvolveCdcTable).isTrue();
+ }
+
+ private TableId registerTable(String keyspaceName, String tableName,
boolean cdc)
+ {
+ CassandraBridgeImplementation.setup();
+ KeyspaceMetadata keyspaceMetadata =
KeyspaceMetadata.create(keyspaceName, KeyspaceParams.simple(1));
+ Schema.instance.load(keyspaceMetadata);
+ Keyspace.openWithoutSSTables(keyspaceName);
+
+ String createTableStatement = "CREATE TABLE " + keyspaceName + "." +
tableName + " (pk int PRIMARY KEY)"
+ + (cdc ? " WITH cdc = true" : "");
+ TableMetadata tableMetadata = CQLFragmentParser
+
.parseAny(CqlParser::createTableStatement, createTableStatement, "CREATE TABLE")
+ .keyspace(keyspaceName)
+ .prepare(null)
+ .builder(Types.none())
+ .build();
+ KeyspaceMetadata keyspace =
Schema.instance.getKeyspaceMetadata(keyspaceName);
+
Schema.instance.load(keyspace.withSwapped(keyspace.tables.with(tableMetadata)));
+ return tableMetadata.id;
+ }
+
+ // Builds a mutation buffer by hand: a valid header (vint update count,
followed by the real TableId of the
+ // first update) followed by garbage bytes standing in for a partition
update body that fails to deserialize
+ // (e.g. due to a schema mismatch). The checksum in
BufferingCommitLogReader already guarantees these bytes
+ // aren't randomly corrupted, so this reproduces a mutation whose first
TableId is intact but whose body
+ // cannot be parsed.
+ private byte[] buildCorruptedMutation(int updateCount, TableId
firstTableId) throws Exception
+ {
+ try (DataOutputBuffer out = new DataOutputBuffer())
+ {
+ out.writeUnsignedVInt(updateCount);
+ firstTableId.serialize(out);
+ out.write(new byte[]{(byte) 0xDE, (byte) 0xAD, (byte) 0xBE, (byte)
0xEF});
+ return out.toByteArray();
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
