Add version string to v9 transport handshake (#120744)

Author: DaveCTurner

server/src/main/java/org/elasticsearch/transport/TcpTransport.java

@@ -12,6 +12,7 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.apache.lucene.util.BytesRef;
+import org.elasticsearch.Build;
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.TransportVersion;
 import org.elasticsearch.action.ActionListener;
@@ -175,7 +176,7 @@ public TcpTransport(
                 channel,
                 requestId,
                 TransportHandshaker.HANDSHAKE_ACTION_NAME,
-                new TransportHandshaker.HandshakeRequest(version),
+                new TransportHandshaker.HandshakeRequest(version, Build.current().version()),
                 TransportRequestOptions.EMPTY,
                 v,
                 null,

server/src/main/java/org/elasticsearch/transport/TransportHandshaker.java

@@ -9,6 +9,7 @@
 
 package org.elasticsearch.transport;
 
+import org.elasticsearch.Build;
 import org.elasticsearch.TransportVersion;
 import org.elasticsearch.TransportVersions;
 import org.elasticsearch.action.ActionListener;
@@ -19,10 +20,12 @@
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.metrics.CounterMetric;
 import org.elasticsearch.core.TimeValue;
+import org.elasticsearch.core.UpdateForV9;
 import org.elasticsearch.threadpool.ThreadPool;
 
 import java.io.EOFException;
 import java.io.IOException;
+import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
@@ -206,7 +209,7 @@ void handleHandshake(TransportChannel channel, long requestId, StreamInput strea
             assert ignoreDeserializationErrors : exception;
             throw exception;
         }
-        channel.sendResponse(new HandshakeResponse(this.version));
+        channel.sendResponse(new HandshakeResponse(this.version, Build.current().version()));
     }
 
     TransportResponseHandler<HandshakeResponse> removeHandlerForHandshake(long requestId) {
@@ -245,7 +248,7 @@ public Executor executor() {
         @Override
         public void handleResponse(HandshakeResponse response) {
             if (isDone.compareAndSet(false, true)) {
-                TransportVersion responseVersion = response.responseVersion;
+                TransportVersion responseVersion = response.transportVersion;
                 if (TransportVersion.isCompatible(responseVersion) == false) {
                     listener.onFailure(
                         new IllegalStateException(
@@ -257,7 +260,7 @@ public void handleResponse(HandshakeResponse response) {
                         )
                     );
                 } else {
-                    listener.onResponse(TransportVersion.min(TransportHandshaker.this.version, response.getResponseVersion()));
+                    listener.onResponse(TransportVersion.min(TransportHandshaker.this.version, response.getTransportVersion()));
                 }
             }
         }
@@ -278,12 +281,23 @@ void handleLocalException(TransportException e) {
 
     static final class HandshakeRequest extends TransportRequest {
 
-        private final TransportVersion version;
+        /**
+         * The {@link TransportVersion#current()} of the requesting node.
+         */
+        final TransportVersion transportVersion;
 
-        HandshakeRequest(TransportVersion version) {
-            this.version = version;
+        /**
+         * The {@link Build#version()} of the requesting node, as a {@link String}, for better reporting of handshake failures due to
+         * an incompatible version.
+         */
+        final String releaseVersion;
+
+        HandshakeRequest(TransportVersion transportVersion, String releaseVersion) {
+            this.transportVersion = Objects.requireNonNull(transportVersion);
+            this.releaseVersion = Objects.requireNonNull(releaseVersion);
         }
 
+        @UpdateForV9(owner = UpdateForV9.Owner.CORE_INFRA) // remainingMessage == null is invalid in v9
         HandshakeRequest(StreamInput streamInput) throws IOException {
             super(streamInput);
             BytesReference remainingMessage;
@@ -293,53 +307,101 @@ static final class HandshakeRequest extends TransportRequest {
                 remainingMessage = null;
             }
             if (remainingMessage == null) {
-                version = null;
+                transportVersion = null;
+                releaseVersion = null;
             } else {
                 try (StreamInput messageStreamInput = remainingMessage.streamInput()) {
-                    this.version = TransportVersion.readVersion(messageStreamInput);
+                    this.transportVersion = TransportVersion.readVersion(messageStreamInput);
+                    if (streamInput.getTransportVersion().onOrAfter(V9_HANDSHAKE_VERSION)) {
+                        this.releaseVersion = messageStreamInput.readString();
+                    } else {
+                        this.releaseVersion = this.transportVersion.toReleaseVersion();
+                    }
                 }
             }
         }
 
         @Override
         public void writeTo(StreamOutput streamOutput) throws IOException {
             super.writeTo(streamOutput);
-            assert version != null;
-            try (BytesStreamOutput messageStreamOutput = new BytesStreamOutput(4)) {
-                TransportVersion.writeVersion(version, messageStreamOutput);
+            assert transportVersion != null;
+            try (BytesStreamOutput messageStreamOutput = new BytesStreamOutput(1024)) {
+                TransportVersion.writeVersion(transportVersion, messageStreamOutput);
+                if (streamOutput.getTransportVersion().onOrAfter(V9_HANDSHAKE_VERSION)) {
+                    messageStreamOutput.writeString(releaseVersion);
+                } // else we just send the transport version and rely on a best-effort mapping to release versions
                 BytesReference reference = messageStreamOutput.bytes();
                 streamOutput.writeBytesReference(reference);
             }
         }
     }
 
+    /**
+     * A response to a low-level transport handshake, carrying information about the version of the responding node.
+     */
     static final class HandshakeResponse extends TransportResponse {
 
-        private final TransportVersion responseVersion;
+        /**
+         * The {@link TransportVersion#current()} of the responding node.
+         */
+        private final TransportVersion transportVersion;
 
-        HandshakeResponse(TransportVersion responseVersion) {
-            this.responseVersion = responseVersion;
+        /**
+         * The {@link Build#version()} of the responding node, as a {@link String}, for better reporting of handshake failures due to
+         * an incompatible version.
+         */
+        private final String releaseVersion;
+
+        HandshakeResponse(TransportVersion transportVersion, String releaseVersion) {
+            this.transportVersion = Objects.requireNonNull(transportVersion);
+            this.releaseVersion = Objects.requireNonNull(releaseVersion);
         }
 
-        private HandshakeResponse(StreamInput in) throws IOException {
+        HandshakeResponse(StreamInput in) throws IOException {
             super(in);
-            responseVersion = TransportVersion.readVersion(in);
+            transportVersion = TransportVersion.readVersion(in);
+            if (in.getTransportVersion().onOrAfter(V9_HANDSHAKE_VERSION)) {
+                releaseVersion = in.readString();
+            } else {
+                releaseVersion = transportVersion.toReleaseVersion();
+            }
         }
 
         @Override
         public void writeTo(StreamOutput out) throws IOException {
-            assert responseVersion != null;
-            TransportVersion.writeVersion(responseVersion, out);
+            TransportVersion.writeVersion(transportVersion, out);
+            if (out.getTransportVersion().onOrAfter(V9_HANDSHAKE_VERSION)) {
+                out.writeString(releaseVersion);
+            } // else we just send the transport version and rely on a best-effort mapping to release versions
+        }
+
+        /**
+         * @return the {@link TransportVersion#current()} of the responding node.
+         */
+        TransportVersion getTransportVersion() {
+            return transportVersion;
         }
 
-        TransportVersion getResponseVersion() {
-            return responseVersion;
+        /**
+         * @return the {@link Build#version()} of the responding node, as a {@link String}, for better reporting of handshake failures due
+         * to an incompatible version.
+         */
+        String getReleaseVersion() {
+            return releaseVersion;
         }
     }
 
     @FunctionalInterface
     interface HandshakeRequestSender {
-
-        void sendRequest(DiscoveryNode node, TcpChannel channel, long requestId, TransportVersion version) throws IOException;
+        /**
+         * @param node                      The (expected) remote node, for error reporting and passing to
+         *                                  {@link TransportMessageListener#onRequestSent}.
+         * @param channel                   The TCP channel to use to send the handshake request.
+         * @param requestId                 The transport request ID, for matching up the response.
+         * @param handshakeTransportVersion The {@link TransportVersion} to use for the handshake request, which will be
+         *                                  {@link TransportHandshaker#V8_HANDSHAKE_VERSION} in production.
+         */
+        void sendRequest(DiscoveryNode node, TcpChannel channel, long requestId, TransportVersion handshakeTransportVersion)
+            throws IOException;
     }
 }