From 55a9778575b36545aba8f40b563b5280309f07ce Mon Sep 17 00:00:00 2001 From: Brandon Williams Date: Wed, 17 Jun 2026 14:02:19 -0500 Subject: [PATCH 1/8] CNDB-18201: support streaming with mixed MS versions MessagingService.accept_streaming now accepts a range between VERSION_40 and current_version. Inbound streaming uses the negotiated version and outbound streaming stores it as a Netty channel attribute that control/file messages use for serialization. --- .../db/streaming/CassandraIncomingFile.java | 2 +- .../db/streaming/CassandraOutgoingFile.java | 14 +++++-- .../db/streaming/CassandraStreamReader.java | 9 ++++- .../db/streaming/CassandraStreamWriter.java | 11 ++++-- .../net/InboundConnectionInitiator.java | 10 ++--- .../cassandra/net/MessagingService.java | 2 +- .../streaming/DefaultConnectionFactory.java | 7 +++- .../async/NettyStreamingMessageSender.java | 17 +++++++-- .../async/StreamCompressionSerializer.java | 2 - .../async/StreamingInboundHandler.java | 1 + .../apache/cassandra/net/HandshakeTest.java | 37 +++++++++++++++++++ 11 files changed, 91 insertions(+), 21 deletions(-) diff --git a/src/java/org/apache/cassandra/db/streaming/CassandraIncomingFile.java b/src/java/org/apache/cassandra/db/streaming/CassandraIncomingFile.java index 2d6a4fdc7354..bc6f309744b4 100644 --- a/src/java/org/apache/cassandra/db/streaming/CassandraIncomingFile.java +++ b/src/java/org/apache/cassandra/db/streaming/CassandraIncomingFile.java @@ -78,7 +78,7 @@ public synchronized void read(DataInputPlus in, int version) throws IOException else if (streamHeader.isCompressed()) reader = new CassandraCompressedStreamReader(header, streamHeader, session); else - reader = new CassandraStreamReader(header, streamHeader, session); + reader = new CassandraStreamReader(header, streamHeader, session, version); size = streamHeader.size(); sstable = reader.read(in); diff --git a/src/java/org/apache/cassandra/db/streaming/CassandraOutgoingFile.java b/src/java/org/apache/cassandra/db/streaming/CassandraOutgoingFile.java index 0e089c9d0096..52de5f6047c4 100644 --- a/src/java/org/apache/cassandra/db/streaming/CassandraOutgoingFile.java +++ b/src/java/org/apache/cassandra/db/streaming/CassandraOutgoingFile.java @@ -178,10 +178,16 @@ public void write(StreamSession session, DataOutputStreamPlus out, int version) CassandraStreamHeader.serializer.serialize(header, out, version); out.flush(); - CassandraStreamWriter writer = header.isCompressed() ? - new CassandraCompressedStreamWriter(sstable, header, session) : - new CassandraStreamWriter(sstable, header, session); - writer.write(out); + if (header.isCompressed()) + { + CassandraCompressedStreamWriter writer = new CassandraCompressedStreamWriter(sstable, header, session); + writer.write(out); + } + else + { + CassandraStreamWriter writer = new CassandraStreamWriter(sstable, header, session); + writer.write(out, version); + } } } diff --git a/src/java/org/apache/cassandra/db/streaming/CassandraStreamReader.java b/src/java/org/apache/cassandra/db/streaming/CassandraStreamReader.java index 3a8a9b363f79..a8753855e65b 100644 --- a/src/java/org/apache/cassandra/db/streaming/CassandraStreamReader.java +++ b/src/java/org/apache/cassandra/db/streaming/CassandraStreamReader.java @@ -79,8 +79,14 @@ public class CassandraStreamReader implements IStreamReader protected final int sstableLevel; protected final SerializationHeader.Component header; protected final int fileSeqNum; + protected final int protocolVersion; public CassandraStreamReader(StreamMessageHeader header, CassandraStreamHeader streamHeader, StreamSession session) + { + this(header, streamHeader, session, current_version); + } + + public CassandraStreamReader(StreamMessageHeader header, CassandraStreamHeader streamHeader, StreamSession session, int protocolVersion) { if (session.getPendingRepair() != null) { @@ -99,6 +105,7 @@ public CassandraStreamReader(StreamMessageHeader header, CassandraStreamHeader s this.sstableLevel = streamHeader.sstableLevel; this.header = streamHeader.serializationHeader; this.fileSeqNum = header.sequenceNumber; + this.protocolVersion = protocolVersion; } /** @@ -125,7 +132,7 @@ public SSTableMultiWriter read(DataInputPlus inputPlus) throws IOException StreamDeserializer deserializer = null; SSTableMultiWriter writer = null; - try (StreamCompressionInputStream streamCompressionInputStream = new StreamCompressionInputStream(inputPlus, current_version)) + try (StreamCompressionInputStream streamCompressionInputStream = new StreamCompressionInputStream(inputPlus, protocolVersion)) { TrackedDataInputPlus in = new TrackedDataInputPlus(streamCompressionInputStream); deserializer = new StreamDeserializer(cfs.metadata(), in, inputVersion, getHeader(cfs.metadata())); diff --git a/src/java/org/apache/cassandra/db/streaming/CassandraStreamWriter.java b/src/java/org/apache/cassandra/db/streaming/CassandraStreamWriter.java index 180ae8f25e41..ae10a756a7f0 100644 --- a/src/java/org/apache/cassandra/db/streaming/CassandraStreamWriter.java +++ b/src/java/org/apache/cassandra/db/streaming/CassandraStreamWriter.java @@ -78,6 +78,11 @@ public CassandraStreamWriter(SSTableReader sstable, CassandraStreamHeader header * @throws IOException on any I/O error */ public void write(DataOutputStreamPlus output) throws IOException + { + write(output, current_version); + } + + public void write(DataOutputStreamPlus output, int version) throws IOException { long totalSize = totalSize(); logger.debug("[Stream #{}] Start streaming file {} to {}, repairedAt = {}, totalSize = {}", session.planId(), @@ -110,7 +115,7 @@ public void write(DataOutputStreamPlus output) throws IOException while (bytesRead < length) { int toTransfer = (int) Math.min(bufferSize, length - bytesRead); - long lastBytesRead = write(proxy, validator, out, start, transferOffset, toTransfer, bufferSize); + long lastBytesRead = write(proxy, validator, out, start, transferOffset, toTransfer, bufferSize, version); start += lastBytesRead; bytesRead += lastBytesRead; progress += (lastBytesRead - transferOffset); @@ -144,7 +149,7 @@ protected long totalSize() * * @throws java.io.IOException on any I/O error */ - protected long write(ChannelProxy proxy, ChecksumValidator validator, AsyncStreamingOutputPlus output, long start, int transferOffset, int toTransfer, int bufferSize) throws IOException + protected long write(ChannelProxy proxy, ChecksumValidator validator, AsyncStreamingOutputPlus output, long start, int transferOffset, int toTransfer, int bufferSize, int version) throws IOException { // the count of bytes to read off disk int minReadable = (int) Math.min(bufferSize, proxy.size() - (start - sstable.getDataFileSliceDescriptor().sliceStart)); @@ -166,7 +171,7 @@ protected long write(ChannelProxy proxy, ChecksumValidator validator, AsyncStrea buffer.position(transferOffset); buffer.limit(transferOffset + (toTransfer - transferOffset)); - output.writeToChannel(StreamCompressionSerializer.serialize(compressor, buffer, current_version), limiter); + output.writeToChannel(StreamCompressionSerializer.serialize(compressor, buffer, version), limiter); } finally { diff --git a/src/java/org/apache/cassandra/net/InboundConnectionInitiator.java b/src/java/org/apache/cassandra/net/InboundConnectionInitiator.java index cbfb3c6931c9..186eb2d2ef5f 100644 --- a/src/java/org/apache/cassandra/net/InboundConnectionInitiator.java +++ b/src/java/org/apache/cassandra/net/InboundConnectionInitiator.java @@ -322,7 +322,7 @@ else if (initiate.acceptVersions.max < accept.min) } else if (initiate.type.isStreaming()) { - setupStreamingPipeline(initiate.from, ctx); + setupStreamingPipeline(initiate.from, useMessagingVersion, ctx); } else { @@ -348,7 +348,7 @@ else if (initiate.type.isStreaming()) logger.warn("Received stream using protocol version {} (my version {}). Terminating connection", version, settings.acceptStreaming.max); failHandshake(ctx); } - setupStreamingPipeline(initiate.from, ctx); + setupStreamingPipeline(initiate.from, version, ctx); } else { @@ -448,7 +448,7 @@ private void failHandshake(Channel channel) handshakeTimeout.cancel(true); } - private void setupStreamingPipeline(InetAddressAndPort from, ChannelHandlerContext ctx) + private void setupStreamingPipeline(InetAddressAndPort from, int streamingVersion, ChannelHandlerContext ctx) { handshakeTimeout.cancel(true); assert initiate.framing == Framing.UNPROTECTED; @@ -463,7 +463,7 @@ private void setupStreamingPipeline(InetAddressAndPort from, ChannelHandlerConte } BufferPools.forNetworking().setRecycleWhenFreeForCurrentThread(false); - pipeline.replace(this, "streamInbound", new StreamingInboundHandler(from, current_version, null)); + pipeline.replace(this, "streamInbound", new StreamingInboundHandler(from, streamingVersion, null)); logger.info("{} streaming connection established, version = {}, framing = {}, encryption = {}", SocketFactory.channelId(from, @@ -472,7 +472,7 @@ private void setupStreamingPipeline(InetAddressAndPort from, ChannelHandlerConte (InetSocketAddress) channel.localAddress(), ConnectionType.STREAMING, channel.id().asShortText()), - current_version, + streamingVersion, initiate.framing, SocketFactory.encryptionConnectionSummary(pipeline.channel())); } diff --git a/src/java/org/apache/cassandra/net/MessagingService.java b/src/java/org/apache/cassandra/net/MessagingService.java index 3f45063ed4b7..64aa48b05b5c 100644 --- a/src/java/org/apache/cassandra/net/MessagingService.java +++ b/src/java/org/apache/cassandra/net/MessagingService.java @@ -226,7 +226,7 @@ public class MessagingService extends MessagingServiceMBeanImpl public static final int VERSION_DSE_68 = 168; static AcceptVersions accept_messaging = new AcceptVersions(minimum_version, current_version, SUPPORTED_DSE_VERSION); - static AcceptVersions accept_streaming = new AcceptVersions(current_version, current_version); + static AcceptVersions accept_streaming = new AcceptVersions(Math.min(VERSION_40, current_version), current_version); static Map versionOrdinalMap = Arrays.stream(Version.values()).collect(Collectors.toMap(v -> v.value, Enum::ordinal)); @Deprecated // remove when cndb no longer supports bdp/6.8-cndb diff --git a/src/java/org/apache/cassandra/streaming/DefaultConnectionFactory.java b/src/java/org/apache/cassandra/streaming/DefaultConnectionFactory.java index 5f2163f5410b..e98c7e3c47b5 100644 --- a/src/java/org/apache/cassandra/streaming/DefaultConnectionFactory.java +++ b/src/java/org/apache/cassandra/streaming/DefaultConnectionFactory.java @@ -30,6 +30,7 @@ import org.apache.cassandra.net.OutboundConnectionInitiator.Result; import org.apache.cassandra.net.OutboundConnectionInitiator.Result.StreamingSuccess; import org.apache.cassandra.net.OutboundConnectionSettings; +import org.apache.cassandra.streaming.async.NettyStreamingMessageSender; import static org.apache.cassandra.net.OutboundConnectionInitiator.initiateStreaming; @@ -49,7 +50,11 @@ public Channel createConnection(OutboundConnectionSettings template, int messagi Future> result = initiateStreaming(eventLoop, template.withDefaults(ConnectionCategory.STREAMING), messagingVersion); result.awaitUninterruptibly(); // initiate has its own timeout, so this is "guaranteed" to return relatively promptly if (result.isSuccess()) - return result.getNow().success().channel; + { + StreamingSuccess success = result.getNow().success(); + success.channel.attr(NettyStreamingMessageSender.STREAMING_VERSION_ATTR).set(success.messagingVersion); + return success.channel; + } if (++attempts == MAX_CONNECT_ATTEMPTS) throw new IOException("failed to connect to " + template.to + " for streaming data", result.cause()); diff --git a/src/java/org/apache/cassandra/streaming/async/NettyStreamingMessageSender.java b/src/java/org/apache/cassandra/streaming/async/NettyStreamingMessageSender.java index 9f8f47608d9f..d51babc66dec 100644 --- a/src/java/org/apache/cassandra/streaming/async/NettyStreamingMessageSender.java +++ b/src/java/org/apache/cassandra/streaming/async/NettyStreamingMessageSender.java @@ -123,6 +123,8 @@ public class NettyStreamingMessageSender implements StreamingMessageSender @VisibleForTesting static final AttributeKey TRANSFERRING_FILE_ATTR = AttributeKey.valueOf("transferringFile"); + public static final AttributeKey STREAMING_VERSION_ATTR = AttributeKey.valueOf("streamingVersion"); + public NettyStreamingMessageSender(StreamSession session, OutboundConnectionSettings template, StreamConnectionFactory factory, int streamingVersion, boolean isPreview) { this.session = session; @@ -161,6 +163,7 @@ public void injectControlMessageChannel(Channel channel) { this.controlMessageChannel = channel; channel.attr(TRANSFERRING_FILE_ATTR).set(Boolean.FALSE); + channel.attr(STREAMING_VERSION_ATTR).compareAndSet(null, streamingVersion); scheduleKeepAliveTask(channel); } @@ -197,6 +200,7 @@ private void scheduleKeepAliveTask(Channel channel) private Channel createChannel(boolean isInboundHandlerNeeded, OutboundConnectionSettings templateWithConnectTo) throws IOException { Channel channel = factory.createConnection(templateWithConnectTo, streamingVersion); + channel.attr(STREAMING_VERSION_ATTR).compareAndSet(null, streamingVersion); session.attachOutbound(channel); if (isInboundHandlerNeeded) @@ -261,7 +265,8 @@ private void sendControlMessage(Channel channel, StreamMessage message, GenericF logger.debug("{} Sending {}", createLogTag(session, channel), message); // we anticipate that the control messages are rather small, so allocating a ByteBuf shouldn't blow out of memory. - long messageSize = StreamMessage.serializedSize(message, streamingVersion); + int channelStreamingVersion = streamingVersion(channel); + long messageSize = StreamMessage.serializedSize(message, channelStreamingVersion); if (messageSize > 1 << 30) { throw new IllegalStateException(String.format("%s something is seriously wrong with the calculated stream control message's size: %d bytes, type is %s", @@ -273,7 +278,7 @@ private void sendControlMessage(Channel channel, StreamMessage message, GenericF ByteBuffer nioBuf = buf.nioBuffer(0, (int) messageSize); @SuppressWarnings("resource") DataOutputBufferFixed out = new DataOutputBufferFixed(nioBuf); - StreamMessage.serialize(message, out, streamingVersion, session); + StreamMessage.serialize(message, out, channelStreamingVersion, session); assert nioBuf.position() == nioBuf.limit(); buf.writerIndex(nioBuf.position()); @@ -351,7 +356,7 @@ public void run() // close the DataOutputStreamPlus as we're done with it - but don't close the channel try (DataOutputStreamPlus outPlus = new AsyncStreamingOutputPlus(channel)) { - StreamMessage.serialize(msg, outPlus, streamingVersion, session); + StreamMessage.serialize(msg, outPlus, streamingVersion(channel), session); } finally { @@ -543,6 +548,12 @@ int semaphoreAvailablePermits() return fileTransferSemaphore.availablePermits(); } + private int streamingVersion(Channel channel) + { + Integer channelStreamingVersion = channel.attr(STREAMING_VERSION_ATTR).get(); + return channelStreamingVersion == null ? streamingVersion : channelStreamingVersion; + } + @Override public boolean connected() { diff --git a/src/java/org/apache/cassandra/streaming/async/StreamCompressionSerializer.java b/src/java/org/apache/cassandra/streaming/async/StreamCompressionSerializer.java index fc1bde2e5b10..0b757d61c35a 100644 --- a/src/java/org/apache/cassandra/streaming/async/StreamCompressionSerializer.java +++ b/src/java/org/apache/cassandra/streaming/async/StreamCompressionSerializer.java @@ -29,7 +29,6 @@ import org.apache.cassandra.io.util.DataInputPlus; import org.apache.cassandra.net.AsyncStreamingOutputPlus; -import static org.apache.cassandra.net.MessagingService.current_version; /** * A serialiazer for stream compressed files (see package-level documentation). Much like a typical compressed @@ -56,7 +55,6 @@ public StreamCompressionSerializer(ByteBufAllocator allocator) public static AsyncStreamingOutputPlus.Write serialize(LZ4Compressor compressor, ByteBuffer in, int version) { - assert version == current_version; return bufferSupplier -> { int uncompressedLength = in.remaining(); int maxLength = compressor.maxCompressedLength(uncompressedLength); diff --git a/src/java/org/apache/cassandra/streaming/async/StreamingInboundHandler.java b/src/java/org/apache/cassandra/streaming/async/StreamingInboundHandler.java index 3b9c1723858e..13b7fd4ddf0a 100644 --- a/src/java/org/apache/cassandra/streaming/async/StreamingInboundHandler.java +++ b/src/java/org/apache/cassandra/streaming/async/StreamingInboundHandler.java @@ -88,6 +88,7 @@ public StreamingInboundHandler(InetAddressAndPort remoteAddress, int protocolVer public void handlerAdded(ChannelHandlerContext ctx) { buffers = new AsyncStreamingInputPlus(ctx.channel()); + ctx.channel().attr(NettyStreamingMessageSender.STREAMING_VERSION_ATTR).set(protocolVersion); Thread blockingIOThread = new FastThreadLocalThread(new StreamDeserializingTask(session, ctx.channel()), String.format("Stream-Deserializer-%s-%s", remoteAddress.toString(), ctx.channel().id())); blockingIOThread.setDaemon(true); diff --git a/test/unit/org/apache/cassandra/net/HandshakeTest.java b/test/unit/org/apache/cassandra/net/HandshakeTest.java index 7230b0155ef7..69a98c46344d 100644 --- a/test/unit/org/apache/cassandra/net/HandshakeTest.java +++ b/test/unit/org/apache/cassandra/net/HandshakeTest.java @@ -40,11 +40,13 @@ import org.apache.cassandra.locator.InetAddressAndPort; import org.apache.cassandra.net.OutboundConnectionInitiator.Result; import org.apache.cassandra.net.OutboundConnectionInitiator.Result.MessagingSuccess; +import org.apache.cassandra.net.OutboundConnectionInitiator.Result.StreamingSuccess; import static org.apache.cassandra.net.MessagingService.VERSION_30; import static org.apache.cassandra.net.MessagingService.VERSION_3014; import static org.apache.cassandra.net.MessagingService.VERSION_40; import static org.apache.cassandra.net.MessagingService.VERSION_DS_11; +import static org.apache.cassandra.net.MessagingService.VERSION_DS_12; import static org.apache.cassandra.net.MessagingService.current_version; import static org.apache.cassandra.net.MessagingService.minimum_version; import static org.apache.cassandra.net.ConnectionType.SMALL_MESSAGES; @@ -103,6 +105,41 @@ private Result handshake(int req, AcceptVersions acceptOutbound, AcceptVersions } } + private Result streamingHandshake(int req, int outMin, int outMax, int inMin, int inMax) throws ExecutionException, InterruptedException + { + InboundSockets inbound = new InboundSockets(new InboundConnectionSettings().withAcceptStreaming(new AcceptVersions(inMin, inMax))); + try + { + inbound.open(); + InetAddressAndPort endpoint = inbound.sockets().stream().map(s -> s.settings.bindAddress).findFirst().get(); + EventLoop eventLoop = factory.defaultGroup().next(); + Future> future = + initiateStreaming(eventLoop, + new OutboundConnectionSettings(endpoint) + .withAcceptVersions(new AcceptVersions(outMin, outMax)) + .withDefaults(ConnectionCategory.STREAMING), + req); + return future.get(20, TimeUnit.SECONDS); + } + catch (TimeoutException e) + { + throw new RuntimeException(e); + } + finally + { + inbound.close().await(1L, TimeUnit.SECONDS); + } + } + + @Test + public void testStreamingNegotiatesCommonVersion() throws InterruptedException, ExecutionException + { + Result result = streamingHandshake(VERSION_DS_12, VERSION_40, VERSION_DS_12, VERSION_40, VERSION_DS_11); + Assert.assertEquals(Result.Outcome.SUCCESS, result.outcome); + Assert.assertEquals(VERSION_DS_11, result.success().messagingVersion); + result.success().channel.close(); + } + @Test public void testBothCurrentVersion() throws InterruptedException, ExecutionException { From 0746d1cf81de32a27d5b0116379d0891aade608a Mon Sep 17 00:00:00 2001 From: Brandon Williams Date: Thu, 18 Jun 2026 12:46:34 -0500 Subject: [PATCH 2/8] apply version matrix to streaming tests --- .../cassandra/net/MessagingService.java | 12 ++++ .../streaming/CassandraStreamHeaderTest.java | 21 +++++-- .../db/streaming/StreamRequestTest.java | 27 +++++---- .../StreamCompressionSerializerTest.java | 60 ++++++++++++++----- 4 files changed, 88 insertions(+), 32 deletions(-) diff --git a/src/java/org/apache/cassandra/net/MessagingService.java b/src/java/org/apache/cassandra/net/MessagingService.java index 64aa48b05b5c..9db679100315 100644 --- a/src/java/org/apache/cassandra/net/MessagingService.java +++ b/src/java/org/apache/cassandra/net/MessagingService.java @@ -20,6 +20,7 @@ import java.nio.channels.ClosedChannelException; import java.util.ArrayList; import java.util.Arrays; +import java.util.Collections; import java.util.HashSet; import java.util.List; import java.util.Map; @@ -278,6 +279,17 @@ public enum Version { this.value = value; } + + public static List supportedVersions() + { + List versions = new ArrayList<>(); + for (Version version : values()) + { + if (minimum_version <= version.value) + versions.add(version); + } + return Collections.unmodifiableList(versions); + } } private static class MSHandle diff --git a/test/unit/org/apache/cassandra/db/streaming/CassandraStreamHeaderTest.java b/test/unit/org/apache/cassandra/db/streaming/CassandraStreamHeaderTest.java index 89a9b14d6ef7..85671ee21119 100644 --- a/test/unit/org/apache/cassandra/db/streaming/CassandraStreamHeaderTest.java +++ b/test/unit/org/apache/cassandra/db/streaming/CassandraStreamHeaderTest.java @@ -40,8 +40,10 @@ import org.apache.cassandra.io.sstable.Component; import org.apache.cassandra.io.sstable.format.SSTableFormat; import org.apache.cassandra.io.sstable.format.SSTableReader; +import org.apache.cassandra.io.IVersionedSerializer; import org.apache.cassandra.io.sstable.format.big.BigFormat; import org.apache.cassandra.io.util.DataInputPlus; +import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.schema.CompressionParams; import org.apache.cassandra.schema.KeyspaceParams; import org.apache.cassandra.schema.TableMetadata; @@ -99,7 +101,7 @@ public void transferedSizeWithCompressionTest() header.compressionInfo.chunks(); assertEquals(transferedSize, header.calculateSize()); - SerializationUtils.assertSerializationCycle(header, CassandraStreamHeader.serializer); + assertSerializationCycleForStreamingVersions(header, CassandraStreamHeader.serializer); } @Test @@ -115,7 +117,7 @@ public void transferedSizeWithZeroCopyStreamingTest() header.compressionInfo.chunks(); assertEquals(transferedSize, header.calculateSize()); - SerializationUtils.assertSerializationCycle(header, CassandraStreamHeader.serializer); + assertSerializationCycleForStreamingVersions(header, CassandraStreamHeader.serializer); } @Test @@ -128,7 +130,7 @@ public void transferedSizeWithoutCompressionTest() assertEquals(sstable.uncompressedLength(), transferedSize); assertEquals(transferedSize, header.calculateSize()); - SerializationUtils.assertSerializationCycle(header, CassandraStreamHeader.serializer); + assertSerializationCycleForStreamingVersions(header, CassandraStreamHeader.serializer); } private CassandraStreamHeader header(boolean entireSSTable, boolean compressed) @@ -175,7 +177,7 @@ public void serializerTest() .withTableId(metadata.id) .build(); - SerializationUtils.assertSerializationCycle(header, CassandraStreamHeader.serializer); + assertSerializationCycleForStreamingVersions(header, CassandraStreamHeader.serializer); } @Test @@ -200,7 +202,16 @@ public void serializerTest_EntireSSTableTransfer() .withTableId(metadata.id) .build(); - SerializationUtils.assertSerializationCycle(header, new TestableCassandraStreamHeaderSerializer()); + assertSerializationCycleForStreamingVersions(header, new TestableCassandraStreamHeaderSerializer()); + } + + private static void assertSerializationCycleForStreamingVersions(CassandraStreamHeader header, IVersionedSerializer serializer) + { + for (MessagingService.Version version : MessagingService.Version.supportedVersions()) + { + if (version.value >= MessagingService.VERSION_40 && version.value <= MessagingService.current_version) + SerializationUtils.assertSerializationCycle(header, serializer, version.value); + } } private static class TestableCassandraStreamHeaderSerializer extends CassandraStreamHeaderSerializer diff --git a/test/unit/org/apache/cassandra/db/streaming/StreamRequestTest.java b/test/unit/org/apache/cassandra/db/streaming/StreamRequestTest.java index 9f5e65685022..2fbd5941586f 100644 --- a/test/unit/org/apache/cassandra/db/streaming/StreamRequestTest.java +++ b/test/unit/org/apache/cassandra/db/streaming/StreamRequestTest.java @@ -43,7 +43,6 @@ public class StreamRequestTest { private static InetAddressAndPort local; private final String ks = "keyspace"; - private final int version = MessagingService.current_version; @BeforeClass public static void setUp() throws Throwable @@ -62,19 +61,25 @@ public void serializationRoundTrip() throws Throwable Arrays.asList(range(5, 6), range(7, 8))), Arrays.asList("a", "b", "c")); - int expectedSize = (int) StreamRequest.serializer.serializedSize(orig, version); - try (DataOutputBuffer out = new DataOutputBuffer(expectedSize)) + for (MessagingService.Version version : MessagingService.Version.supportedVersions()) { - StreamRequest.serializer.serialize(orig, out, version); - Assert.assertEquals(expectedSize, out.buffer().limit()); - try (DataInputBuffer in = new DataInputBuffer(out.buffer(), false)) + if (version.value < MessagingService.VERSION_40 || version.value > MessagingService.current_version) + continue; + + int expectedSize = (int) StreamRequest.serializer.serializedSize(orig, version.value); + try (DataOutputBuffer out = new DataOutputBuffer(expectedSize)) { - StreamRequest decoded = StreamRequest.serializer.deserialize(in, version); + StreamRequest.serializer.serialize(orig, out, version.value); + Assert.assertEquals(expectedSize, out.buffer().limit()); + try (DataInputBuffer in = new DataInputBuffer(out.buffer(), false)) + { + StreamRequest decoded = StreamRequest.serializer.deserialize(in, version.value); - Assert.assertEquals(orig.keyspace, decoded.keyspace); - Util.assertRCEquals(orig.full, decoded.full); - Util.assertRCEquals(orig.transientReplicas, decoded.transientReplicas); - Assert.assertEquals(orig.columnFamilies, decoded.columnFamilies); + Assert.assertEquals(orig.keyspace, decoded.keyspace); + Util.assertRCEquals(orig.full, decoded.full); + Util.assertRCEquals(orig.transientReplicas, decoded.transientReplicas); + Assert.assertEquals(orig.columnFamilies, decoded.columnFamilies); + } } } } diff --git a/test/unit/org/apache/cassandra/streaming/async/StreamCompressionSerializerTest.java b/test/unit/org/apache/cassandra/streaming/async/StreamCompressionSerializerTest.java index dab600111e17..12e69117a9f6 100644 --- a/test/unit/org/apache/cassandra/streaming/async/StreamCompressionSerializerTest.java +++ b/test/unit/org/apache/cassandra/streaming/async/StreamCompressionSerializerTest.java @@ -42,7 +42,6 @@ public class StreamCompressionSerializerTest { - private static final int VERSION = MessagingService.current_version; private static final Random random = new Random(2347623847623L); private final ByteBufAllocator allocator = PooledByteBufAllocator.DEFAULT; @@ -62,23 +61,44 @@ public static void before() @After public void tearDown() + { + releaseBuffers(); + } + + private void releaseBuffers() { if (input != null) + { FileUtils.clean(input); + input = null; + } if (compressed != null) + { FileUtils.clean(compressed); + compressed = null; + } if (output != null && output.refCnt() > 0) + { output.release(output.refCnt()); + output = null; + } } @Test public void roundTrip_HappyPath_NotReadabaleByteBuffer() throws IOException { - populateInput(); - StreamCompressionSerializer.serialize(compressor, input, VERSION).write(size -> compressed = ByteBuffer.allocateDirect(size)); - input.flip(); - output = serializer.deserialize(decompressor, new DataInputBuffer(compressed, false), VERSION); - validateResults(); + for (MessagingService.Version version : MessagingService.Version.supportedVersions()) + { + if (!isStreamingVersion(version.value)) + continue; + + releaseBuffers(); + populateInput(); + StreamCompressionSerializer.serialize(compressor, input, version.value).write(size -> compressed = ByteBuffer.allocateDirect(size)); + input.flip(); + output = serializer.deserialize(decompressor, new DataInputBuffer(compressed, false), version.value); + validateResults(); + } } private void populateInput() @@ -100,16 +120,24 @@ private void validateResults() @Test public void roundTrip_HappyPath_ReadabaleByteBuffer() throws IOException { - populateInput(); - StreamCompressionSerializer.serialize(compressor, input, VERSION) - .write(size -> { - if (compressed != null) - FileUtils.clean(compressed); - return compressed = ByteBuffer.allocateDirect(size); - }); - input.flip(); - output = serializer.deserialize(decompressor, new ByteBufRCH(Unpooled.wrappedBuffer(compressed)), VERSION); - validateResults(); + for (MessagingService.Version version : MessagingService.Version.supportedVersions()) + { + if (!isStreamingVersion(version.value)) + continue; + + releaseBuffers(); + populateInput(); + StreamCompressionSerializer.serialize(compressor, input, version.value) + .write(size -> compressed = ByteBuffer.allocateDirect(size)); + input.flip(); + output = serializer.deserialize(decompressor, new ByteBufRCH(Unpooled.wrappedBuffer(compressed)), version.value); + validateResults(); + } + } + + private static boolean isStreamingVersion(int version) + { + return version >= MessagingService.VERSION_40 && version <= MessagingService.current_version; } private static class ByteBufRCH extends DataInputBuffer implements ReadableByteChannel From b6e41e10be86075bc409b5cdfd8f03942cfbb8ab Mon Sep 17 00:00:00 2001 From: Brandon Williams Date: Fri, 19 Jun 2026 10:02:58 -0500 Subject: [PATCH 3/8] Add distributed streaming test between DS_11 and DS_12 --- .../distributed/test/StreamingTest.java | 76 ++++++++++++++++++- 1 file changed, 75 insertions(+), 1 deletion(-) diff --git a/test/distributed/org/apache/cassandra/distributed/test/StreamingTest.java b/test/distributed/org/apache/cassandra/distributed/test/StreamingTest.java index 8fff004ad14d..e3effb996aa1 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/StreamingTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/StreamingTest.java @@ -34,9 +34,11 @@ import org.junit.Assert; import org.junit.Test; +import org.apache.cassandra.config.CassandraRelevantProperties; import org.apache.cassandra.distributed.Cluster; import org.apache.cassandra.distributed.api.IInvokableInstance; import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.service.StorageService; import org.apache.cassandra.streaming.StreamSession; import org.apache.cassandra.streaming.messages.StreamMessage; @@ -82,7 +84,7 @@ private void testStreaming(int nodes, int replicationFactor, int rowCount, Strin cluster.get(nodes).runOnInstance(() -> StorageService.instance.rebuild(null, KEYSPACE, null, null)); { Object[][] results = cluster.get(nodes).executeInternal(String.format("SELECT k, c1, c2 FROM %s.cf;", KEYSPACE)); - Assert.assertEquals(1000, results.length); + Assert.assertEquals(rowCount, results.length); Arrays.sort(results, Comparator.comparingInt(a -> Integer.parseInt((String) a[0]))); for (int i = 0 ; i < results.length ; ++i) { @@ -100,6 +102,78 @@ public void test() throws Throwable testStreaming(2, 2, 1000, "LeveledCompactionStrategy"); } + @Test + public void testMixedMessagingVersionStreamingDs11ToDs12() throws Throwable + { + testMixedMessagingVersionStreaming(MessagingService.VERSION_DS_11, MessagingService.VERSION_DS_12); + } + + @Test + public void testMixedMessagingVersionStreamingDs12ToDs11() throws Throwable + { + testMixedMessagingVersionStreaming(MessagingService.VERSION_DS_12, MessagingService.VERSION_DS_11); + } + + private void testMixedMessagingVersionStreaming(int sourceVersion, int targetVersion) throws Throwable + { + int rowCount = 1000; + String keyspace = KEYSPACE + '_' + sourceVersion + '_' + targetVersion; + try (Cluster cluster = builder().withNodes(2) + .withDataDirCount(1) + .withConfig(config -> config.with(NETWORK)) + .withInstanceInitializer((classLoader, node) -> initializeMessagingVersion(classLoader, node == 1 ? sourceVersion : targetVersion)) + .start()) + { + Assert.assertEquals(sourceVersion, (int) cluster.get(1).callOnInstance(() -> MessagingService.current_version)); + Assert.assertEquals(targetVersion, (int) cluster.get(2).callOnInstance(() -> MessagingService.current_version)); + + int schemaCoordinator = sourceVersion <= targetVersion ? 1 : 2; + cluster.schemaChange("CREATE KEYSPACE " + keyspace + " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 2};", false, cluster.get(schemaCoordinator)); + cluster.schemaChange(String.format("CREATE TABLE %s.cf (k text, c1 text, c2 text, PRIMARY KEY (k)) WITH compaction = {'class': 'LeveledCompactionStrategy', 'enabled': 'true'}", keyspace), false, cluster.get(schemaCoordinator)); + + for (int i = 0 ; i < rowCount ; ++i) + cluster.get(1).executeInternal(String.format("INSERT INTO %s.cf (k, c1, c2) VALUES (?, 'value1', 'value2');", keyspace), Integer.toString(i)); + + cluster.get(2).executeInternal("TRUNCATE system.available_ranges;"); + Assert.assertEquals(0, cluster.get(2).executeInternal(String.format("SELECT k, c1, c2 FROM %s.cf;", keyspace)).length); + + registerSink(cluster, 2); + cluster.get(2).runOnInstance(() -> StorageService.instance.rebuild(null, keyspace, null, null)); + + Object[][] results = cluster.get(2).executeInternal(String.format("SELECT k, c1, c2 FROM %s.cf;", keyspace)); + Assert.assertEquals(rowCount, results.length); + Arrays.sort(results, Comparator.comparingInt(a -> Integer.parseInt((String) a[0]))); + for (int i = 0 ; i < results.length ; ++i) + { + Assert.assertEquals(Integer.toString(i), results[i][0]); + Assert.assertEquals("value1", results[i][1]); + Assert.assertEquals("value2", results[i][2]); + } + } + } + + private static void initializeMessagingVersion(ClassLoader classLoader, int version) + { + String key = CassandraRelevantProperties.DS_CURRENT_MESSAGING_VERSION.getKey(); + String previous = System.getProperty(key); + try + { + System.setProperty(key, Integer.toString(version)); + Class.forName(MessagingService.class.getName(), true, classLoader).getField("current_version").getInt(null); + } + catch (ReflectiveOperationException e) + { + throw new RuntimeException(e); + } + finally + { + if (previous == null) + System.clearProperty(key); + else + System.setProperty(key, previous); + } + } + public static void registerSink(Cluster cluster, int initiatorNodeId) { IInvokableInstance initiatorNode = cluster.get(initiatorNodeId); From f7a67b83013cfcb0f3def8009ad29c02d77f699a Mon Sep 17 00:00:00 2001 From: Brandon Williams Date: Fri, 19 Jun 2026 10:38:30 -0500 Subject: [PATCH 4/8] Add uncompressed distributed test variant --- .../distributed/test/StreamingTest.java | 27 ++++++++++++++----- 1 file changed, 21 insertions(+), 6 deletions(-) diff --git a/test/distributed/org/apache/cassandra/distributed/test/StreamingTest.java b/test/distributed/org/apache/cassandra/distributed/test/StreamingTest.java index e3effb996aa1..6b1f546874fe 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/StreamingTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/StreamingTest.java @@ -105,31 +105,46 @@ public void test() throws Throwable @Test public void testMixedMessagingVersionStreamingDs11ToDs12() throws Throwable { - testMixedMessagingVersionStreaming(MessagingService.VERSION_DS_11, MessagingService.VERSION_DS_12); + testMixedMessagingVersionStreaming(MessagingService.VERSION_DS_11, MessagingService.VERSION_DS_12, false); } @Test public void testMixedMessagingVersionStreamingDs12ToDs11() throws Throwable { - testMixedMessagingVersionStreaming(MessagingService.VERSION_DS_12, MessagingService.VERSION_DS_11); + testMixedMessagingVersionStreaming(MessagingService.VERSION_DS_12, MessagingService.VERSION_DS_11, false); } - private void testMixedMessagingVersionStreaming(int sourceVersion, int targetVersion) throws Throwable + @Test + public void testMixedMessagingVersionUncompressedStreamingDs11ToDs12() throws Throwable + { + testMixedMessagingVersionStreaming(MessagingService.VERSION_DS_11, MessagingService.VERSION_DS_12, true); + } + + @Test + public void testMixedMessagingVersionUncompressedStreamingDs12ToDs11() throws Throwable + { + testMixedMessagingVersionStreaming(MessagingService.VERSION_DS_12, MessagingService.VERSION_DS_11, true); + } + + private void testMixedMessagingVersionStreaming(int sourceVersion, int targetVersion, boolean uncompressed) throws Throwable { int rowCount = 1000; - String keyspace = KEYSPACE + '_' + sourceVersion + '_' + targetVersion; + String keyspace = KEYSPACE + '_' + sourceVersion + '_' + targetVersion + (uncompressed ? "_uncompressed" : ""); try (Cluster cluster = builder().withNodes(2) .withDataDirCount(1) - .withConfig(config -> config.with(NETWORK)) + // disable entire sstable streaming so the per-section CassandraStreamReader/Writer path is used + .withConfig(config -> config.with(NETWORK).set("stream_entire_sstables", !uncompressed)) .withInstanceInitializer((classLoader, node) -> initializeMessagingVersion(classLoader, node == 1 ? sourceVersion : targetVersion)) .start()) { Assert.assertEquals(sourceVersion, (int) cluster.get(1).callOnInstance(() -> MessagingService.current_version)); Assert.assertEquals(targetVersion, (int) cluster.get(2).callOnInstance(() -> MessagingService.current_version)); + // disable sstable compression so the uncompressed stream path is exercised + String compression = uncompressed ? " AND compression = {'enabled': 'false'}" : ""; int schemaCoordinator = sourceVersion <= targetVersion ? 1 : 2; cluster.schemaChange("CREATE KEYSPACE " + keyspace + " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 2};", false, cluster.get(schemaCoordinator)); - cluster.schemaChange(String.format("CREATE TABLE %s.cf (k text, c1 text, c2 text, PRIMARY KEY (k)) WITH compaction = {'class': 'LeveledCompactionStrategy', 'enabled': 'true'}", keyspace), false, cluster.get(schemaCoordinator)); + cluster.schemaChange(String.format("CREATE TABLE %s.cf (k text, c1 text, c2 text, PRIMARY KEY (k)) WITH compaction = {'class': 'LeveledCompactionStrategy', 'enabled': 'true'}%s", keyspace, compression), false, cluster.get(schemaCoordinator)); for (int i = 0 ; i < rowCount ; ++i) cluster.get(1).executeInternal(String.format("INSERT INTO %s.cf (k, c1, c2) VALUES (?, 'value1', 'value2');", keyspace), Integer.toString(i)); From 5dd254b50614d304e2403c9f0c86202d48ea360f Mon Sep 17 00:00:00 2001 From: Brandon Williams Date: Fri, 19 Jun 2026 11:55:18 -0500 Subject: [PATCH 5/8] guard against future versions, improve handshake test --- .../net/InboundConnectionInitiator.java | 1 + .../apache/cassandra/net/MessagingService.java | 5 ++++- .../apache/cassandra/net/HandshakeTest.java | 18 ++++++++++++++++++ 3 files changed, 23 insertions(+), 1 deletion(-) diff --git a/src/java/org/apache/cassandra/net/InboundConnectionInitiator.java b/src/java/org/apache/cassandra/net/InboundConnectionInitiator.java index 186eb2d2ef5f..18c748188867 100644 --- a/src/java/org/apache/cassandra/net/InboundConnectionInitiator.java +++ b/src/java/org/apache/cassandra/net/InboundConnectionInitiator.java @@ -347,6 +347,7 @@ else if (initiate.type.isStreaming()) { logger.warn("Received stream using protocol version {} (my version {}). Terminating connection", version, settings.acceptStreaming.max); failHandshake(ctx); + return; } setupStreamingPipeline(initiate.from, version, ctx); } diff --git a/src/java/org/apache/cassandra/net/MessagingService.java b/src/java/org/apache/cassandra/net/MessagingService.java index 9db679100315..d8176ae087ce 100644 --- a/src/java/org/apache/cassandra/net/MessagingService.java +++ b/src/java/org/apache/cassandra/net/MessagingService.java @@ -227,6 +227,9 @@ public class MessagingService extends MessagingServiceMBeanImpl public static final int VERSION_DSE_68 = 168; static AcceptVersions accept_messaging = new AcceptVersions(minimum_version, current_version, SUPPORTED_DSE_VERSION); + // Streaming negotiates any common version in [VERSION_40, current_version]; this is only safe because every + // streaming serializer (stream headers, control/file messages, compressed payloads) is wire-compatible across + // that range. static AcceptVersions accept_streaming = new AcceptVersions(Math.min(VERSION_40, current_version), current_version); static Map versionOrdinalMap = Arrays.stream(Version.values()).collect(Collectors.toMap(v -> v.value, Enum::ordinal)); @@ -285,7 +288,7 @@ public static List supportedVersions() List versions = new ArrayList<>(); for (Version version : values()) { - if (minimum_version <= version.value) + if (minimum_version <= version.value && version.value <= current_version) versions.add(version); } return Collections.unmodifiableList(versions); diff --git a/test/unit/org/apache/cassandra/net/HandshakeTest.java b/test/unit/org/apache/cassandra/net/HandshakeTest.java index 69a98c46344d..dd290cea2c9f 100644 --- a/test/unit/org/apache/cassandra/net/HandshakeTest.java +++ b/test/unit/org/apache/cassandra/net/HandshakeTest.java @@ -140,6 +140,24 @@ public void testStreamingNegotiatesCommonVersion() throws InterruptedException, result.success().channel.close(); } + @Test + public void testStreamingNegotiatesCommonVersionReversed() throws InterruptedException, ExecutionException + { + // lower initiator (max DS_11) to higher follower (max DS_12) still negotiates the common DS_11 + Result result = streamingHandshake(VERSION_DS_11, VERSION_40, VERSION_DS_11, VERSION_40, VERSION_DS_12); + Assert.assertEquals(Result.Outcome.SUCCESS, result.outcome); + Assert.assertEquals(VERSION_DS_11, result.success().messagingVersion); + result.success().channel.close(); + } + + @Test + public void testStreamingIncompatibleVersions() throws InterruptedException, ExecutionException + { + // initiator only supports DS_12, follower only supports up to DS_11: no common version + Result result = streamingHandshake(VERSION_DS_12, VERSION_DS_12, VERSION_DS_12, VERSION_40, VERSION_DS_11); + Assert.assertEquals(Result.Outcome.INCOMPATIBLE, result.outcome); + } + @Test public void testBothCurrentVersion() throws InterruptedException, ExecutionException { From 1307a0714b7eb209ae7e66929bab38a83b006063 Mon Sep 17 00:00:00 2001 From: Brandon Williams Date: Tue, 23 Jun 2026 13:11:54 -0500 Subject: [PATCH 6/8] address review feedback --- .../net/InboundConnectionSettings.java | 2 +- .../cassandra/net/MessagingService.java | 1 + .../db/streaming/StreamRequestTest.java | 3 -- .../apache/cassandra/net/HandshakeTest.java | 29 ++++++++++++++----- 4 files changed, 24 insertions(+), 11 deletions(-) diff --git a/src/java/org/apache/cassandra/net/InboundConnectionSettings.java b/src/java/org/apache/cassandra/net/InboundConnectionSettings.java index 00def4fc47db..003e9ef4733c 100644 --- a/src/java/org/apache/cassandra/net/InboundConnectionSettings.java +++ b/src/java/org/apache/cassandra/net/InboundConnectionSettings.java @@ -131,7 +131,7 @@ public InboundConnectionSettings withAcceptMessaging(AcceptVersions acceptMessag acceptMessaging, acceptStreaming, socketFactory, handlers); } - public InboundConnectionSettings withAcceptStreaming(AcceptVersions acceptMessaging) + public InboundConnectionSettings withAcceptStreaming(AcceptVersions acceptStreaming) { return new InboundConnectionSettings(authenticator, bindAddress, encryption, socketReceiveBufferSizeInBytes, applicationReceiveQueueCapacityInBytes, diff --git a/src/java/org/apache/cassandra/net/MessagingService.java b/src/java/org/apache/cassandra/net/MessagingService.java index d8176ae087ce..f2abeb83e2a9 100644 --- a/src/java/org/apache/cassandra/net/MessagingService.java +++ b/src/java/org/apache/cassandra/net/MessagingService.java @@ -283,6 +283,7 @@ public enum Version this.value = value; } + @VisibleForTesting public static List supportedVersions() { List versions = new ArrayList<>(); diff --git a/test/unit/org/apache/cassandra/db/streaming/StreamRequestTest.java b/test/unit/org/apache/cassandra/db/streaming/StreamRequestTest.java index 2fbd5941586f..91a69b1238b0 100644 --- a/test/unit/org/apache/cassandra/db/streaming/StreamRequestTest.java +++ b/test/unit/org/apache/cassandra/db/streaming/StreamRequestTest.java @@ -63,9 +63,6 @@ public void serializationRoundTrip() throws Throwable for (MessagingService.Version version : MessagingService.Version.supportedVersions()) { - if (version.value < MessagingService.VERSION_40 || version.value > MessagingService.current_version) - continue; - int expectedSize = (int) StreamRequest.serializer.serializedSize(orig, version.value); try (DataOutputBuffer out = new DataOutputBuffer(expectedSize)) { diff --git a/test/unit/org/apache/cassandra/net/HandshakeTest.java b/test/unit/org/apache/cassandra/net/HandshakeTest.java index dd290cea2c9f..79936836ea2a 100644 --- a/test/unit/org/apache/cassandra/net/HandshakeTest.java +++ b/test/unit/org/apache/cassandra/net/HandshakeTest.java @@ -19,6 +19,7 @@ package org.apache.cassandra.net; import java.nio.channels.ClosedChannelException; +import java.util.List; import java.util.Objects; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; @@ -131,30 +132,44 @@ private Result streamingHandshake(int req, int outMin, int outMax, int inMin, in } } + private static int latestStreamingVersionBeforeCurrent() + { + List versions = MessagingService.Version.supportedVersions(); + int previousVersion = -1; + for (MessagingService.Version version : versions) + { + if (version.value >= VERSION_40 && version.value < current_version) + previousVersion = version.value; + } + Assert.assertTrue(previousVersion >= VERSION_40); + return previousVersion; + } + @Test public void testStreamingNegotiatesCommonVersion() throws InterruptedException, ExecutionException { - Result result = streamingHandshake(VERSION_DS_12, VERSION_40, VERSION_DS_12, VERSION_40, VERSION_DS_11); + int previousVersion = latestStreamingVersionBeforeCurrent(); + Result result = streamingHandshake(current_version, VERSION_40, current_version, VERSION_40, previousVersion); Assert.assertEquals(Result.Outcome.SUCCESS, result.outcome); - Assert.assertEquals(VERSION_DS_11, result.success().messagingVersion); + Assert.assertEquals(previousVersion, result.success().messagingVersion); result.success().channel.close(); } @Test public void testStreamingNegotiatesCommonVersionReversed() throws InterruptedException, ExecutionException { - // lower initiator (max DS_11) to higher follower (max DS_12) still negotiates the common DS_11 - Result result = streamingHandshake(VERSION_DS_11, VERSION_40, VERSION_DS_11, VERSION_40, VERSION_DS_12); + int previousVersion = latestStreamingVersionBeforeCurrent(); + Result result = streamingHandshake(previousVersion, VERSION_40, previousVersion, VERSION_40, current_version); Assert.assertEquals(Result.Outcome.SUCCESS, result.outcome); - Assert.assertEquals(VERSION_DS_11, result.success().messagingVersion); + Assert.assertEquals(previousVersion, result.success().messagingVersion); result.success().channel.close(); } @Test public void testStreamingIncompatibleVersions() throws InterruptedException, ExecutionException { - // initiator only supports DS_12, follower only supports up to DS_11: no common version - Result result = streamingHandshake(VERSION_DS_12, VERSION_DS_12, VERSION_DS_12, VERSION_40, VERSION_DS_11); + int previousVersion = latestStreamingVersionBeforeCurrent(); + Result result = streamingHandshake(current_version, current_version, current_version, VERSION_40, previousVersion); Assert.assertEquals(Result.Outcome.INCOMPATIBLE, result.outcome); } From 2116e88f30c7d0544199dffff6417b3a945481a7 Mon Sep 17 00:00:00 2001 From: Brandon Williams Date: Wed, 24 Jun 2026 13:57:14 -0500 Subject: [PATCH 7/8] Update test/distributed/org/apache/cassandra/distributed/test/StreamingTest.java MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Andrés de la Peña --- .../org/apache/cassandra/distributed/test/StreamingTest.java | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/test/distributed/org/apache/cassandra/distributed/test/StreamingTest.java b/test/distributed/org/apache/cassandra/distributed/test/StreamingTest.java index 6b1f546874fe..0fc533d10151 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/StreamingTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/StreamingTest.java @@ -44,6 +44,7 @@ import org.apache.cassandra.streaming.messages.StreamMessage; import static org.apache.cassandra.distributed.api.Feature.NETWORK; +import static org.apache.cassandra.distributed.shared.AssertUtils.assertRow; import static org.apache.cassandra.streaming.StreamSession.State.PREPARING; import static org.apache.cassandra.streaming.StreamSession.State.STREAMING; import static org.apache.cassandra.streaming.StreamSession.State.WAIT_COMPLETE; @@ -160,9 +161,7 @@ private void testMixedMessagingVersionStreaming(int sourceVersion, int targetVer Arrays.sort(results, Comparator.comparingInt(a -> Integer.parseInt((String) a[0]))); for (int i = 0 ; i < results.length ; ++i) { - Assert.assertEquals(Integer.toString(i), results[i][0]); - Assert.assertEquals("value1", results[i][1]); - Assert.assertEquals("value2", results[i][2]); + assertRow(results[i], Integer.toString(i), "value1", "value2"); } } } From 795d09aef66adb632c2e3cc68c8a0b0d05635696 Mon Sep 17 00:00:00 2001 From: Brandon Williams Date: Wed, 24 Jun 2026 13:59:16 -0500 Subject: [PATCH 8/8] remove unused import --- test/unit/org/apache/cassandra/net/HandshakeTest.java | 1 - 1 file changed, 1 deletion(-) diff --git a/test/unit/org/apache/cassandra/net/HandshakeTest.java b/test/unit/org/apache/cassandra/net/HandshakeTest.java index 79936836ea2a..0214d5a9f187 100644 --- a/test/unit/org/apache/cassandra/net/HandshakeTest.java +++ b/test/unit/org/apache/cassandra/net/HandshakeTest.java @@ -47,7 +47,6 @@ import static org.apache.cassandra.net.MessagingService.VERSION_3014; import static org.apache.cassandra.net.MessagingService.VERSION_40; import static org.apache.cassandra.net.MessagingService.VERSION_DS_11; -import static org.apache.cassandra.net.MessagingService.VERSION_DS_12; import static org.apache.cassandra.net.MessagingService.current_version; import static org.apache.cassandra.net.MessagingService.minimum_version; import static org.apache.cassandra.net.ConnectionType.SMALL_MESSAGES;