You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by ma...@apache.org on 2026/10/21 13:47:13 UTC
(cassandra) 01/01: Merge branch 'cassandra-4.0' into cassandra-4.1
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 <ma...@apache.org>
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: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org