CASSANALYTICS-196: Skip CDC-irrelevant mutations on deserialization failure

CHANGES.txt

@@ -1,5 +1,6 @@
 0.5.0
 -----
+ * Skip CDC-irrelevant mutations on deserialization failure (CASSANALYTICS-196)
  * Eliminate redundant filesystem lookups in SSTable direct streaming (CASSANALYTICS-104)
  * TokenPartitioner fails to detect range gap in reader (CASSANALYTICS-180)
  * CDC reader stats silently dropped in SidecarCdcBuilder (CASSANALYTICS-191)

cassandra-analytics-common/src/main/java/org/apache/cassandra/cdc/stats/ICdcStats.java

@@ -179,6 +179,16 @@ default void mutationsDeserializeFailedCount(long incrCount)
     {
     }
 
+    /**
+     * 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
      *

cassandra-four-zero-bridge/src/main/java/org/apache/cassandra/db/commitlog/BufferingCommitLogReader.java

@@ -51,6 +51,9 @@
 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;
@@ -567,6 +570,16 @@ private void readMutationInternal(byte[] inputBuffer,
         catch (Throwable t)
         {
             JVMStabilityInspector.inspectThrowable(t);
+
+            if (failedMutationHasNoCdcTable(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)))
@@ -604,6 +617,45 @@ private void readMutationInternal(byte[] inputBuffer,
         this.handleMutation(mutation, size, mutationPosition, desc);
     }
 
+    /**
+     * Best-effort check for whether a mutation that failed to deserialize 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 false 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); true otherwise, including
+     *         when the table's metadata can't be found, or even the first {@link TableId} can't be read
+     */
+    @VisibleForTesting
+    static boolean failedMutationHasNoCdcTable(byte[] inputBuffer, int size)
+    {
+        try (DataInputBuffer in = new DataInputBuffer(inputBuffer, 0, size))
+        {
+            int updateCount = (int) in.readUnsignedVInt();
+            TableId tableId = TableId.deserialize(in);
+            TableMetadata metadata = Schema.instance.getTableMetadata(tableId);
+            if (metadata != null && metadata.params.cdc)
+            {
+                return false;
+            }
+
+            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 false;
+            }
+        }
+        catch (Throwable t)
+        {
+            // unable to read the first TableId
+        }
+        return true;
+    }
+
     public void close()
     {
         if (updates == null)

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 testFailedMutationHasNoCdcTableWhenNonCdcTableIdReadableButRestCorrupted() throws Exception
+    {
+        TableId tableId = registerTable("ks_non_cdc", "tbl", false);
+        byte[] corruptedMutation = buildCorruptedMutation(1, tableId);
+
+        boolean failedMutationHasNoCdcTable = BufferingCommitLogReader.failedMutationHasNoCdcTable(corruptedMutation,
+                                                                                                     corruptedMutation.length);
+
+        assertThat(failedMutationHasNoCdcTable).isTrue();
+    }
+
+    @Test
+    public void testFailedMutationHasNoCdcTableWhenCdcTableIdReadableButRestCorrupted() throws Exception
+    {
+        TableId tableId = registerTable("ks_cdc", "tbl", true);
+        byte[] corruptedMutation = buildCorruptedMutation(1, tableId);
+
+        boolean failedMutationHasNoCdcTable = BufferingCommitLogReader.failedMutationHasNoCdcTable(corruptedMutation,
+                                                                                                     corruptedMutation.length);
+
+        assertThat(failedMutationHasNoCdcTable).isFalse();
+    }
+
+    @Test
+    public void testFailedMutationHasNoCdcTableWhenTableIdUnresolvable() throws Exception
+    {
+        CassandraBridgeImplementation.setup();
+        TableId unknownTableId = TableId.fromUUID(UUID.randomUUID());
+        byte[] corruptedMutation = buildCorruptedMutation(1, unknownTableId);
+
+        boolean failedMutationHasNoCdcTable = BufferingCommitLogReader.failedMutationHasNoCdcTable(corruptedMutation,
+                                                                                                     corruptedMutation.length);
+
+        // an unresolvable table is treated as not CDC-enabled
+        assertThat(failedMutationHasNoCdcTable).isTrue();
+    }
+
+    @Test
+    public void testFailedMutationHasNoCdcTableWhenMultipleUpdatesEvenIfFirstIsNonCdc() throws Exception
+    {
+        TableId tableId = registerTable("ks_non_cdc_multi", "tbl", false);
+        byte[] corruptedMutation = buildCorruptedMutation(2, tableId);
+
+        boolean failedMutationHasNoCdcTable = BufferingCommitLogReader.failedMutationHasNoCdcTable(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(failedMutationHasNoCdcTable).isFalse();
+    }
+
+    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();
+        }
+    }
+}