Make the version of the remote node accessible on a transport channel (#23019)

This commit adds a new method to the TransportChannel that provides access to the version of the
remote node that the response is being sent on and that the request came from. This is helpful
for serialization of data attached as headers.

core/src/main/java/org/elasticsearch/transport/DelegatingTransportChannel.java

@@ -19,6 +19,8 @@
 
 package org.elasticsearch.transport;
 
+import org.elasticsearch.Version;
+
 import java.io.IOException;
 
 /**
@@ -71,4 +73,9 @@ public class DelegatingTransportChannel implements TransportChannel {
     public TransportChannel getChannel() {
         return channel;
     }
+
+    @Override
+    public Version getVersion() {
+        return channel.getVersion();
+    }
 }

core/src/main/java/org/elasticsearch/transport/TcpTransportChannel.java

@@ -105,5 +105,9 @@ public final class TcpTransportChannel<Channel> implements TransportChannel {
         return channel;
     }
 
+    @Override
+    public Version getVersion() {
+        return version;
+    }
 }
 

core/src/main/java/org/elasticsearch/transport/TransportChannel.java

@@ -19,6 +19,8 @@
 
 package org.elasticsearch.transport;
 
+import org.elasticsearch.Version;
+
 import java.io.IOException;
 
 /**
@@ -39,4 +41,11 @@ public interface TransportChannel {
     void sendResponse(TransportResponse response, TransportResponseOptions options) throws IOException;
 
     void sendResponse(Exception exception) throws IOException;
+
+    /**
+     * Returns the version of the other party that this channel will send a response to.
+     */
+    default Version getVersion() {
+        return Version.CURRENT;
+    }
 }

core/src/main/java/org/elasticsearch/transport/TransportService.java

@@ -1155,6 +1155,11 @@ public class TransportService extends AbstractLifecycleComponent {
         public String getChannelType() {
             return "direct";
         }
+
+        @Override
+        public Version getVersion() {
+            return localNode.getVersion();
+        }
     }
 
     /**

modules/transport-netty4/src/test/java/org/elasticsearch/transport/netty4/SimpleNetty4TransportTests.java

@@ -72,8 +72,10 @@ public class SimpleNetty4TransportTests extends AbstractSimpleTransportTestCase
                 return version;
             }
         };
-        return new MockTransportService(Settings.EMPTY, transport, threadPool, TransportService.NOOP_TRANSPORT_INTERCEPTOR,
-                clusterSettings);
+        MockTransportService mockTransportService =
+            MockTransportService.createNewService(Settings.EMPTY, transport, version, threadPool, clusterSettings);
+        mockTransportService.start();
+        return mockTransportService;
     }
 
     @Override

test/framework/src/main/java/org/elasticsearch/transport/AbstractSimpleTransportTestCase.java

@@ -1248,6 +1248,7 @@ public abstract class AbstractSimpleTransportTestCase extends ESTestCase {
                     response.value1 = 1;
                     response.value2 = 2;
                     channel.sendResponse(response);
+                    assertEquals(version0, channel.getVersion());
                 }
             });
 
@@ -1289,6 +1290,7 @@ public abstract class AbstractSimpleTransportTestCase extends ESTestCase {
                     Version0Response response = new Version0Response();
                     response.value1 = 1;
                     channel.sendResponse(response);
+                    assertEquals(version0, channel.getVersion());
                 }
             });
 
@@ -1333,6 +1335,7 @@ public abstract class AbstractSimpleTransportTestCase extends ESTestCase {
                 response.value1 = 1;
                 response.value2 = 2;
                 channel.sendResponse(response);
+                assertEquals(version1, channel.getVersion());
             });
 
         Version1Request version1Request = new Version1Request();
@@ -1374,6 +1377,7 @@ public abstract class AbstractSimpleTransportTestCase extends ESTestCase {
                 Version0Response response = new Version0Response();
                 response.value1 = 1;
                 channel.sendResponse(response);
+                assertEquals(version0, channel.getVersion());
             });
 
         Version0Request version0Request = new Version0Request();