CNDB-18201: support streaming with mixed MS versions

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);

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);
+            }
         }
     }
 

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()));

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
         {

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
                     {
@@ -347,8 +347,9 @@ 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, ctx);
+                    setupStreamingPipeline(initiate.from, version, ctx);
                 }
                 else
                 {
@@ -448,7 +449,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 +464,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 +473,7 @@ private void setupStreamingPipeline(InetAddressAndPort from, ChannelHandlerConte
                                                 (InetSocketAddress) channel.localAddress(),
                                                 ConnectionType.STREAMING,
                                                 channel.id().asShortText()),
-                        current_version,
+                        streamingVersion,
                         initiate.framing,
                         SocketFactory.encryptionConnectionSummary(pipeline.channel()));
         }

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,

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;
@@ -226,7 +227,10 @@ 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);
+    // 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<Integer, Integer> versionOrdinalMap = Arrays.stream(Version.values()).collect(Collectors.toMap(v -> v.value, Enum::ordinal));
 
     @Deprecated // remove when cndb no longer supports bdp/6.8-cndb
@@ -278,6 +282,18 @@ public enum Version
         {
             this.value = value;
         }
+
+        @VisibleForTesting
+        public static List<Version> supportedVersions()
+        {
+            List<Version> versions = new ArrayList<>();
+            for (Version version : values())
+            {
+                if (minimum_version <= version.value && version.value <= current_version)
+                    versions.add(version);
+            }
+            return Collections.unmodifiableList(versions);
+        }
     }
 
     private static class MSHandle

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<StreamingSuccess>> 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());

src/java/org/apache/cassandra/streaming/async/NettyStreamingMessageSender.java

@@ -123,6 +123,8 @@ public class NettyStreamingMessageSender implements StreamingMessageSender
     @VisibleForTesting
     static final AttributeKey<Boolean> TRANSFERRING_FILE_ATTR = AttributeKey.valueOf("transferringFile");
 
+    public static final AttributeKey<Integer> 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()
     {

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);

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);