diff --git a/CHANGES.txt b/CHANGES.txt index 3992e6e77..975ffdf19 100644 --- a/CHANGES.txt +++ b/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) 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 3098ce4cd..2e26bff6f 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 @@ 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 * 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 312cb96c3..a06452f13 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.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) 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 000000000..3ee4f8b34 --- /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 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(); + } + } +}