This is an automated email from the ASF dual-hosted git repository. asf-gitbox-commits pushed a commit to branch cassandra-4.1 in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit 2bebf8f0b891ea6d74e2447d50077177399483f4 Merge: fe284a3af9 80578c2df5 Author: Marcus Eriksson <[email protected]> AuthorDate: Wed Oct 21 15:22:05 2026 +0200 Merge branch 'cassandra-4.0' into cassandra-4.1 CHANGES.txt | 1 + conf/cassandra.yaml | 8 ++ src/java/org/apache/cassandra/config/Config.java | 6 + .../cassandra/config/DatabaseDescriptor.java | 11 ++ .../CassandraEntireSSTableStreamReader.java | 38 ++++++ .../apache/cassandra/metrics/StreamingMetrics.java | 8 ++ .../apache/cassandra/streaming/StreamSession.java | 5 + .../test/hostreplacement/HostReplacementTest.java | 83 +++++++++++++ .../CassandraEntireSSTableStreamWriterTest.java | 138 +++++++++++++++++++++ 9 files changed, 298 insertions(+) diff --cc CHANGES.txt index 785454b15b,7fcbc24dce..c6dbd85e4f --- a/CHANGES.txt +++ b/CHANGES.txt @@@ -1,5 -1,5 +1,6 @@@ -4.0.22 +4.1.13 +Merged from 4.0: + * Validate Digest.crc32 when zero-copy streaming SSTables (CASSANDRA-21694) * Avoid DNS lookup for hostname column in ClientsTable and GossipInfoTable (CASSANDRA-21539) * Separate memtable memory usage tracking from limit checking, and fixes a a flush writeBarrier deadlock (CASSANDRA-21019) * Fixes for fully-expired SSTable index notification in compaction (CASSANDRA-21577) diff --cc conf/cassandra.yaml index 734c938c84,e7bb1c65ad..1e613fc0d8 --- a/conf/cassandra.yaml +++ b/conf/cassandra.yaml @@@ -1074,17 -942,14 +1074,25 @@@ uuid_sstable_identifiers_enabled: fals # disabled if internode encryption is enabled. # stream_entire_sstables: true + # When an entire SSTable is received via zero-copy streaming, validate the Digest.crc32 that was streamed with it + # against the received Data.db before the SSTable is added to the table. This detects corruption introduced in + # transit as well as pre-existing corruption on the sending node's disk, neither of which is otherwise noticed by + # this path, since it writes component bytes to disk without parsing them. + # If there's no Digest.crc32, this check passes. + # When unset, the default is not validating Digest.crc32. + # entire_sstable_stream_digest_validation_enabled: false + +# Throttles entire SSTable outbound streaming file transfers on +# this node to the given total throughput in Mbps. +# Setting this value to 0 it disables throttling. +# When unset, the default is 200 Mbps or 24 MiB/s. +# entire_sstable_stream_throughput_outbound: 24MiB/s + +# Throttles entire SSTable file streaming between datacenters. +# Setting this value to 0 disables throttling for entire SSTable inter-DC file streaming. +# When unset, the default is 200 Mbps or 24 MiB/s. +# entire_sstable_inter_dc_stream_throughput_outbound: 24MiB/s + # Throttles all outbound streaming file transfers on this node to the # given total throughput in Mbps. This is necessary because Cassandra does # mostly sequential IO when streaming data during bootstrap or repair, which diff --cc src/java/org/apache/cassandra/db/streaming/CassandraEntireSSTableStreamReader.java index 515c85dea6,dadc026f3a..c876926158 --- a/src/java/org/apache/cassandra/db/streaming/CassandraEntireSSTableStreamReader.java +++ b/src/java/org/apache/cassandra/db/streaming/CassandraEntireSSTableStreamReader.java @@@ -18,16 -18,21 +18,19 @@@ package org.apache.cassandra.db.streaming; -import java.io.File; import java.io.IOException; import java.util.Collection; + import java.util.List; + import java.util.concurrent.TimeUnit; import java.util.function.UnaryOperator; -import com.google.common.base.Throwables; -import org.apache.cassandra.db.lifecycle.LifecycleNewTracker; import org.slf4j.Logger; import org.slf4j.LoggerFactory; + import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.ColumnFamilyStore; import org.apache.cassandra.db.Directories; +import org.apache.cassandra.db.lifecycle.LifecycleNewTracker; import org.apache.cassandra.io.sstable.Component; import org.apache.cassandra.io.sstable.Descriptor; import org.apache.cassandra.io.sstable.SSTableMultiWriter; @@@ -35,7 -40,7 +38,8 @@@ import org.apache.cassandra.io.sstable. import org.apache.cassandra.io.sstable.format.big.BigTableZeroCopyWriter; import org.apache.cassandra.io.sstable.metadata.StatsMetadata; import org.apache.cassandra.io.util.DataInputPlus; + import org.apache.cassandra.io.util.DataIntegrityMetadata; +import org.apache.cassandra.io.util.File; import org.apache.cassandra.schema.TableId; import org.apache.cassandra.streaming.ProgressInfo; import org.apache.cassandra.streaming.StreamReceiver; diff --cc src/java/org/apache/cassandra/metrics/StreamingMetrics.java index e38b6053ae,0fe1f76f88..8a31227bd3 --- a/src/java/org/apache/cassandra/metrics/StreamingMetrics.java +++ b/src/java/org/apache/cassandra/metrics/StreamingMetrics.java @@@ -47,8 -47,7 +47,9 @@@ public class StreamingMetric public final Counter outgoingBytes; /* Measures the time taken for processing the incoming stream message after being deserialized, including the time to flush to disk. */ public final Timer incomingProcessTime; + private final Counter entireSSTablesStreamedIn; + private final Counter partialSSTablesStreamedIn; + private final Counter entireSSTableDigestMismatches; public static StreamingMetrics get(InetAddressAndPort ip) { @@@ -82,12 -81,12 +83,19 @@@ outgoingBytes= Metrics.counter(factory.createMetricName("OutgoingBytes")); incomingProcessTime = Metrics.timer(factory.createMetricName("IncomingProcessTime")); + entireSSTablesStreamedIn = Metrics.counter(factory.createMetricName("EntireSSTablesStreamedIn")); + partialSSTablesStreamedIn = Metrics.counter(factory.createMetricName("PartialSSTablesStreamedIn")); + entireSSTableDigestMismatches = Metrics.counter(factory.createMetricName("EntireSSTableDigestMismatches")); } + public void countStreamedIn(boolean isEntireSSTable) + { + (isEntireSSTable ? entireSSTablesStreamedIn : partialSSTablesStreamedIn).inc(); + } ++ + /** An entire-sstable stream was rejected because the received data file did not match the streamed digest. */ + public void countEntireSSTableDigestMismatch() + { + entireSSTableDigestMismatches.inc(); + } } diff --cc src/java/org/apache/cassandra/streaming/StreamSession.java index d41fe77abc,0d2668c927..bbd0b33b4b --- a/src/java/org/apache/cassandra/streaming/StreamSession.java +++ b/src/java/org/apache/cassandra/streaming/StreamSession.java @@@ -752,11 -754,12 +752,16 @@@ public class StreamSession implements I }); } + public void countStreamedIn(boolean isEntireSSTable) + { + metrics.countStreamedIn(isEntireSSTable); + } + + public void countEntireSSTableDigestMismatch() + { + metrics.countEntireSSTableDigestMismatch(); + } + - /** * Finish preparing the session. This method is blocking (memtables are flushed in {@link #addTransferRanges}), * so the logic should not execute on the main IO thread (read: netty event loop). diff --cc test/distributed/org/apache/cassandra/distributed/test/hostreplacement/HostReplacementTest.java index 3de0bf51d5,941174f681..fa95a8aebf --- a/test/distributed/org/apache/cassandra/distributed/test/hostreplacement/HostReplacementTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/hostreplacement/HostReplacementTest.java @@@ -205,6 -211,83 +211,83 @@@ public class HostReplacementTest extend } } + private static final int HEALTHY_PK = 10; + private static final int CORRUPT_PK = 11; + + /** + * With digest validation enabled, a single corrupted sstable must fail the entire stream session, so neither + * the corrupted sstable nor its healthy sibling should land on the replacing node. + */ + @Test + public void replaceHostAbortsStreamingWhenDigestValidationCatchesCorruption() throws IOException, TimeoutException + { + TokenSupplier even = TokenSupplier.evenlyDistributedTokens(2); + try (Cluster cluster = Cluster.build(2) + .withConfig(c -> c.with(Feature.GOSSIP, Feature.NETWORK) + .set("entire_sstable_stream_digest_validation_enabled", true)) + .withTokenSupplier(node -> even.token(node == 3 ? 2 : node)) + .start()) + { + IInvokableInstance node1 = cluster.get(1); + IInvokableInstance nodeToRemove = cluster.get(2); + + setupCluster(cluster); + writeHealthyAndCorruptSSTables(cluster, node1); + + stopUnchecked(nodeToRemove); + + IInvokableInstance replacingNode = replaceHostAndStart(cluster, nodeToRemove); + + replacingNode.logs().watchFor("Digest mismatch"); + + // the whole stream session is aborted on a digest mismatch, so neither sstable should have landed + AssertUtils.assertRows(replacingNode.executeInternal("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = " + HEALTHY_PK), EMPTY_ROWS); + AssertUtils.assertRows(replacingNode.executeInternal("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = " + CORRUPT_PK), EMPTY_ROWS); + } + } + + /** + * Flushes two separate sstables on {@code node1} for {@code KEYSPACE.tbl}: one containing {@link #HEALTHY_PK} + * left untouched, and one containing {@link #CORRUPT_PK} that gets its Data.db corrupted. + */ + static void writeHealthyAndCorruptSSTables(Cluster cluster, IInvokableInstance node1) throws IOException + { + node1.nodetoolResult("disableautocompaction").asserts().success(); + + cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + ".tbl (pk) VALUES (?)", + ConsistencyLevel.ALL, + HEALTHY_PK); + node1.flush(KEYSPACE); + + cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + ".tbl (pk) VALUES (?)", + ConsistencyLevel.ALL, + CORRUPT_PK); + node1.flush(KEYSPACE); + + corruptSSTable(node1, "tbl"); + } + + /** Flips a byte in the middle of the Data.db file of the most recently flushed sstable for the given table. */ + static void corruptSSTable(IInvokableInstance instance, String table) throws IOException + { + String dataFilePath = instance.callOnInstance(() -> { + ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(table); + SSTableReader sstable = cfs.getLiveSSTables().stream() + .max(SSTableReader.maxTimestampAscending) + .get(); - return sstable.descriptor.filenameFor(Component.DATA); ++ return sstable.descriptor.fileFor(Component.DATA).absolutePath(); + }); + + try (RandomAccessFile raf = new RandomAccessFile(dataFilePath, "rw")) + { + long corruptAt = raf.length() / 2; + raf.seek(corruptAt); + int b = raf.read(); + raf.seek(corruptAt); + raf.write(b ^ 0xFF); + } + } + static void setupCluster(Cluster cluster) { fixDistributedSchemas(cluster); diff --cc test/unit/org/apache/cassandra/db/streaming/CassandraEntireSSTableStreamWriterTest.java index a65b2ea3a0,43099cfed3..5184b5ae3c --- a/test/unit/org/apache/cassandra/db/streaming/CassandraEntireSSTableStreamWriterTest.java +++ b/test/unit/org/apache/cassandra/db/streaming/CassandraEntireSSTableStreamWriterTest.java @@@ -17,18 -17,26 +17,26 @@@ */ package org.apache.cassandra.db.streaming; -import java.io.FileInputStream; import java.io.IOException; + import java.io.InputStream; import java.nio.ByteBuffer; import java.nio.channels.WritableByteChannel; import java.util.Collection; import java.util.Collections; + import java.util.LinkedHashMap; import java.util.Queue; -import java.util.UUID; ++import com.google.common.io.ByteStreams; ++ +import org.apache.cassandra.Util; + import org.apache.cassandra.io.sstable.Component; import org.apache.cassandra.io.sstable.Descriptor; + import org.junit.AfterClass; import org.junit.BeforeClass; import org.junit.Test; + import org.apache.cassandra.config.DatabaseDescriptor; + -import com.google.common.io.ByteStreams; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.ChannelHandlerContext; @@@ -62,8 -70,9 +70,10 @@@ import org.apache.cassandra.streaming.m import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.utils.FBUtilities; +import static org.apache.cassandra.utils.TimeUUID.Generator.nextTimeUUID; + import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.junit.Assert.assertEquals; + import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; public class CassandraEntireSSTableStreamWriterTest @@@ -170,6 -188,124 +189,125 @@@ } } + @Test + public void testCorruptDataComponentFailsDigestValidation() throws Throwable + { + ByteBuf serializedFile = Unpooled.buffer(8192); + ComponentManifest manifest = writeEntireSSTable(serializedFile); + corruptDataComponent(serializedFile, manifest); + + DatabaseDescriptor.setEntireSSTableStreamDigestValidationEnabled(true); + assertThatThrownBy(() -> readEntireSSTable(serializedFile, manifest)) - .hasCauseInstanceOf(IOException.class) - .hasStackTraceContaining("Digest mismatch"); ++ .isInstanceOf(IOException.class) ++ .hasMessageContaining("Digest mismatch") ++ .hasCauseInstanceOf(IOException.class); + } + + @Test + public void testCorruptDataComponentAcceptedWhenValidationDisabled() throws Throwable + { + ByteBuf serializedFile = Unpooled.buffer(8192); + ComponentManifest manifest = writeEntireSSTable(serializedFile); + corruptDataComponent(serializedFile, manifest); + + DatabaseDescriptor.setEntireSSTableStreamDigestValidationEnabled(false); + + // shouldn't throw an error + SSTableMultiWriter writer = readEntireSSTable(serializedFile, manifest); + assertNotNull(writer); + writer.abort(null); + } + + @Test + public void testMissingDigestIsAcceptedWithoutValidation() throws Throwable + { + DatabaseDescriptor.setEntireSSTableStreamDigestValidationEnabled(true); + + ComponentManifest full; + try (ComponentContext context = ComponentContext.create(descriptor)) + { + full = context.manifest(); + } + + // remove digest component from sstable + LinkedHashMap<Component, Long> withoutDigest = new LinkedHashMap<>(); + for (Component component : full.components()) + { + if (!component.equals(Component.DIGEST)) + withoutDigest.put(component, full.sizeOf(component)); + } + ComponentManifest manifest = new ComponentManifest(withoutDigest); + + // stream from live files + ByteBuf serializedFile = Unpooled.buffer(8192); + for (Component component : manifest.components()) + { - try (InputStream in = new FileInputStream(descriptor.filenameFor(component))) ++ try (InputStream in = descriptor.fileFor(component).newInputStream()) + { + serializedFile.writeBytes(ByteStreams.toByteArray(in)); + } + } + + // shouldn't throw an error + SSTableMultiWriter writer = readEntireSSTable(serializedFile, manifest); + assertNotNull(writer); + writer.abort(null); + } + + private ComponentManifest writeEntireSSTable(ByteBuf serializedFile) throws Exception + { + StreamSession session = setupStreamingSessionForTest(); + EmbeddedChannel channel = createMockNettyChannel(serializedFile); + try (AsyncStreamingOutputPlus out = new AsyncStreamingOutputPlus(channel); + ComponentContext context = ComponentContext.create(descriptor)) + { + new CassandraEntireSSTableStreamWriter(sstable, session, context).write(out); + return context.manifest(); + } + } + + private SSTableMultiWriter readEntireSSTable(ByteBuf serializedFile, ComponentManifest manifest) throws Throwable + { + StreamSession session = setupStreamingSessionForTest(); + InetAddressAndPort peer = FBUtilities.getBroadcastAddressAndPort(); + session.prepareReceiving(new StreamSummary(sstable.metadata().id, 1, 5104)); + + CassandraStreamHeader header = + CassandraStreamHeader.builder() + .withSSTableFormat(sstable.descriptor.formatType) + .withSSTableVersion(sstable.descriptor.version) + .withSSTableLevel(0) + .withEstimatedKeys(sstable.estimatedKeys()) + .withSections(Collections.emptyList()) + .withSerializationHeader(sstable.header.toComponent()) + .withComponentManifest(manifest) + .isEntireSSTable(true) + .withFirstKey(sstable.first) + .withTableId(sstable.metadata().id) + .build(); + + CassandraEntireSSTableStreamReader reader = + new CassandraEntireSSTableStreamReader(new StreamMessageHeader(sstable.metadata().id, peer, session.planId(), false, 0, 0, 0, null), + header, session); + + return reader.read(new DataInputBuffer(serializedFile.nioBuffer(), false)); + } + + /** Flips every bit of one byte in the middle of the Data.db region of an already-serialized stream. */ + private static void corruptDataComponent(ByteBuf serializedFile, ComponentManifest manifest) + { + long dataOffset = 0; + for (Component component : manifest.components()) + { + if (component.equals(Component.DATA)) + break; + dataOffset += manifest.sizeOf(component); + } + + int corruptAt = (int) (dataOffset + manifest.sizeOf(Component.DATA) / 2); + serializedFile.setByte(corruptAt, serializedFile.getByte(corruptAt) ^ 0xff); + } + private EmbeddedChannel createMockNettyChannel(ByteBuf serializedFile) throws Exception { WritableByteChannel wbc = new WritableByteChannel() --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
