[8.x] Add deprecation warning to TransportHandshaker (elastic#123188) (elastic#123627)

JVerwolf · web-flow · commit 6ff283aa21cb · 2025-03-04T15:14:41.000-08:00
This PR adds a deprecation warning log to the TransportHandshaker when connecting with nodes &lt; v8.19.
diff --git a/docs/changelog/123627.yaml b/docs/changelog/123627.yaml
@@ -0,0 +1,19 @@
+pr: 123627
+summary: Add deprecation warning to `TransportHandshaker`
+area: Infra/Core
+type: deprecation
+issues: []
+deprecation:
+  title: Deprecate ability to connect to nodes of versions 8.18 and earlier
+  area: REST API
+  details: >
+    Versions 9.1.0 and later of {es} will not support communication with nodes of versions earlier than 8.19.0,
+    so the ability to connect to nodes of earlier versions is deprecated in this version. This applies both to
+    communication within a cluster and communication across clusters (e.g. for <<modules-cross-cluster-search,{ccs}>> or
+    <<xpack-ccr,{ccr}>>).
+
+    {es} will report in its <<deprecation-logging, deprecation logging>> each time it opens a connection to a node that
+    will not be supported from version 9.1.0 onwards. You must upgrade all your clusters to version 8.19.0 or later
+    before upgrading any of your clusters to 9.1.0 or later.
+  impact: >
+   Upgrade all of your clusters to at least 8.19.0 before upgrading any of them to 9.1.0 or later.
diff --git a/server/src/main/java/org/elasticsearch/transport/TransportHandshaker.java b/server/src/main/java/org/elasticsearch/transport/TransportHandshaker.java
@@ -18,7 +18,10 @@
 import org.elasticsearch.common.io.stream.BytesStreamOutput;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.logging.DeprecationCategory;
+import org.elasticsearch.common.logging.DeprecationLogger;
 import org.elasticsearch.common.metrics.CounterMetric;
+import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.core.Strings;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.logging.LogManager;
@@ -161,6 +164,7 @@ final class TransportHandshaker {
      */
 
     private static final Logger logger = LogManager.getLogger(TransportHandshaker.class);
+    private static final DeprecationLogger deprecationLogger = DeprecationLogger.getLogger(logger.getName());
 
     static final TransportVersion V7_HANDSHAKE_VERSION = TransportVersion.fromId(6_08_00_99);
     static final TransportVersion V8_HANDSHAKE_VERSION = TransportVersion.fromId(7_17_00_99);
@@ -172,6 +176,7 @@ final class TransportHandshaker {
     );
 
     static final String HANDSHAKE_ACTION_NAME = "internal:tcp/handshake";
+    static final TransportVersion V8_19_FIRST_VERSION = TransportVersions.INITIAL_ELASTICSEARCH_8_19;
     private final ConcurrentMap<Long, HandshakeResponseHandler> pendingHandshakes = new ConcurrentHashMap<>();
     private final CounterMetric numHandshakes = new CounterMetric();
 
@@ -249,7 +254,13 @@ void handleHandshake(TransportChannel channel, long requestId, StreamInput strea
         }
         channel.sendResponse(
             new HandshakeResponse(
-                ensureCompatibleVersion(version, handshakeRequest.transportVersion, handshakeRequest.releaseVersion, channel),
+                ensureCompatibleVersion(
+                    version,
+                    handshakeRequest.transportVersion,
+                    handshakeRequest.releaseVersion,
+                    channel,
+                    threadPool.getThreadContext()
+                ),
                 Build.current().version()
             )
         );
@@ -258,10 +269,21 @@ void handleHandshake(TransportChannel channel, long requestId, StreamInput strea
     private static TransportVersion ensureCompatibleVersion(
         TransportVersion localTransportVersion,
         TransportVersion remoteTransportVersion,
-        String releaseVersion,
-        Object channel
+        String remoteReleaseVersion,
+        Object channel,
+        ThreadContext threadContext
     ) {
         if (TransportVersion.isCompatible(remoteTransportVersion)) {
+            // Prevent log message headers from being added to the handshake response.
+            try (var ignored = threadContext.stashContext()) {
+                if (remoteTransportVersion.before(V8_19_FIRST_VERSION)) {
+                    deprecationLogger.warn(
+                        DeprecationCategory.OTHER,
+                        "handshake_version",
+                        getDeprecationMessage(localTransportVersion, remoteTransportVersion, remoteReleaseVersion, channel)
+                    );
+                }
+            }
             if (remoteTransportVersion.onOrAfter(localTransportVersion)) {
                 // Remote is semantically newer than us (i.e. has a greater transport protocol version), so we propose using our current
                 // transport protocol version. If we're initiating the connection then that's the version we'll use; if the other end is
@@ -282,7 +304,7 @@ private static TransportVersion ensureCompatibleVersion(
                             from a chronologically-older release with a numerically-newer version compared to this node's version [{}/{}]. \
                             Upgrading to a chronologically-older release may not work reliably and is not recommended. \
                             Falling back to transport protocol version [{}].""",
-                        releaseVersion,
+                        remoteReleaseVersion,
                         remoteTransportVersion,
                         channel,
                         Build.current().version(),
@@ -298,7 +320,7 @@ private static TransportVersion ensureCompatibleVersion(
             """
                 Rejecting unreadable transport handshake from remote node with version [%s/%s] received on [%s] since this node has \
                 version [%s/%s] which has an incompatible wire format.""",
-            releaseVersion,
+            remoteReleaseVersion,
             remoteTransportVersion,
             channel,
             Build.current().version(),
@@ -309,6 +331,24 @@ private static TransportVersion ensureCompatibleVersion(
 
     }
 
+    // Non-private for testing
+    static String getDeprecationMessage(
+        TransportVersion localTransportVersion,
+        TransportVersion remoteTransportVersion,
+        String remoteReleaseVersion,
+        Object channel
+    ) {
+        return Strings.format(
+            "Performed a handshake with a remote node with version [%s/%s] received on [%s] which "
+                + "will be incompatible after this node on version [%s/%s] is upgraded to >= 9.1.",
+            remoteReleaseVersion,
+            remoteTransportVersion,
+            channel,
+            Build.current().version(),
+            localTransportVersion
+        );
+    }
+
     TransportResponseHandler<HandshakeResponse> removeHandlerForHandshake(long requestId) {
         return pendingHandshakes.remove(requestId);
     }
@@ -352,7 +392,8 @@ public void handleResponse(HandshakeResponse response) {
                         version,
                         response.getTransportVersion(),
                         response.getReleaseVersion(),
-                        channel
+                        channel,
+                        threadPool.getThreadContext()
                     );
                     assert TransportVersion.current().before(version) // simulating a newer-version transport service for test purposes
                         || resultVersion.isKnown() : "negotiated unknown version " + resultVersion;
diff --git a/server/src/test/java/org/elasticsearch/transport/TransportHandshakerTests.java b/server/src/test/java/org/elasticsearch/transport/TransportHandshakerTests.java
@@ -32,6 +32,8 @@
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 
+import static org.elasticsearch.transport.TransportHandshaker.V8_19_FIRST_VERSION;
+import static org.elasticsearch.transport.TransportHandshaker.getDeprecationMessage;
 import static org.hamcrest.Matchers.allOf;
 import static org.hamcrest.Matchers.containsString;
 import static org.mockito.Mockito.doThrow;
@@ -101,10 +103,8 @@ public void testHandshakeRequestAndResponse() throws IOException {
 
     @TestLogging(reason = "testing WARN logging", value = "org.elasticsearch.transport.TransportHandshaker:WARN")
     public void testIncompatibleHandshakeRequest() throws Exception {
-        TransportHandshaker.HandshakeRequest handshakeRequest = new TransportHandshaker.HandshakeRequest(
-            getRandomIncompatibleTransportVersion(),
-            randomIdentifier()
-        );
+        var remoteVersion = getRandomIncompatibleTransportVersion();
+        TransportHandshaker.HandshakeRequest handshakeRequest = new TransportHandshaker.HandshakeRequest(remoteVersion, randomIdentifier());
         BytesStreamOutput bytesStreamOutput = new BytesStreamOutput();
         bytesStreamOutput.setTransportVersion(HANDSHAKE_REQUEST_VERSION);
         handshakeRequest.writeTo(bytesStreamOutput);
@@ -149,7 +149,7 @@ public void testIncompatibleHandshakeRequest() throws Exception {
                 handshakeRequest.transportVersion.bestKnownVersion(),
                 asInstanceOf(TransportHandshaker.HandshakeResponse.class, responseFuture.result()).getTransportVersion()
             );
-
+            assertDeprecationMessageIsLogged(remoteVersion, remoteVersion.toReleaseVersion(), channel);
         } else {
             final TestTransportChannel channel = new TestTransportChannel(ActionListener.running(() -> fail("should not complete")));
 
@@ -186,10 +186,13 @@ public void testHandshakeResponseFromOlderNode() throws Exception {
         assertFalse(versionFuture.isDone());
 
         final var remoteVersion = TransportVersionUtils.randomCompatibleVersion(random());
-        handler.handleResponse(new TransportHandshaker.HandshakeResponse(remoteVersion, randomIdentifier()));
+        var releaseVersion = randomIdentifier();
+        handler.handleResponse(new TransportHandshaker.HandshakeResponse(remoteVersion, releaseVersion));
 
         assertTrue(versionFuture.isDone());
         assertEquals(remoteVersion, versionFuture.result());
+
+        assertDeprecationMessageIsLogged(remoteVersion, releaseVersion, channel);
     }
 
     @TestLogging(reason = "testing WARN logging", value = "org.elasticsearch.transport.TransportHandshaker:WARN")
@@ -201,10 +204,11 @@ public void testHandshakeResponseFromOlderNodeWithPatchedProtocol() throws Excep
 
         assertFalse(versionFuture.isDone());
 
-        final var randomIncompatibleTransportVersion = getRandomIncompatibleTransportVersion();
-        final var handshakeResponse = new TransportHandshaker.HandshakeResponse(randomIncompatibleTransportVersion, randomIdentifier());
+        var remoteVersion = getRandomIncompatibleTransportVersion();
+        var releaseVersion = randomIdentifier();
+        final var handshakeResponse = new TransportHandshaker.HandshakeResponse(remoteVersion, releaseVersion);
 
-        if (randomIncompatibleTransportVersion.onOrAfter(TransportVersions.MINIMUM_COMPATIBLE)) {
+        if (remoteVersion.onOrAfter(TransportVersions.MINIMUM_COMPATIBLE)) {
             // we fall back to the best known version
             MockLog.assertThatLogger(
                 () -> handler.handleResponse(handshakeResponse),
@@ -223,13 +227,13 @@ public void testHandshakeResponseFromOlderNodeWithPatchedProtocol() throws Excep
                         handshakeResponse.getTransportVersion(),
                         Build.current().version(),
                         TransportVersion.current(),
-                        randomIncompatibleTransportVersion.bestKnownVersion()
+                        remoteVersion.bestKnownVersion()
                     )
                 )
             );
 
             assertTrue(versionFuture.isDone());
-            assertEquals(randomIncompatibleTransportVersion.bestKnownVersion(), versionFuture.result());
+            assertEquals(remoteVersion.bestKnownVersion(), versionFuture.result());
         } else {
             MockLog.assertThatLogger(
                 () -> handler.handleResponse(handshakeResponse),
@@ -253,6 +257,13 @@ public void testHandshakeResponseFromOlderNodeWithPatchedProtocol() throws Excep
                 )
             );
         }
+        assertDeprecationMessageIsLogged(remoteVersion, releaseVersion, channel);
+    }
+
+    private void assertDeprecationMessageIsLogged(TransportVersion remoteVersion, String remoteReleaseVersion, Object channel) {
+        if (remoteVersion.onOrAfter(TransportVersions.MINIMUM_COMPATIBLE) && remoteVersion.before(V8_19_FIRST_VERSION)) {
+            assertWarnings(getDeprecationMessage(TransportVersion.current(), remoteVersion, remoteReleaseVersion, channel));
+        }
     }
 
     private static TransportVersion getRandomIncompatibleTransportVersion() {
