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