Remove some dead-code and duplication from o.e.transport (#98321)

original-brownbear · web-flow · commit c203442c4868 · 2023-08-09T21:35:51.000+02:00
Removing some unused things and deduplicating connection lookup failure
handling in TransportService.
diff --git a/modules/transport-netty4/src/main/java/org/elasticsearch/transport/netty4/Netty4MessageInboundHandler.java b/modules/transport-netty4/src/main/java/org/elasticsearch/transport/netty4/Netty4MessageInboundHandler.java
@@ -39,7 +39,6 @@ public Netty4MessageInboundHandler(Netty4Transport transport, Recycler<BytesRef>
         final ThreadPool threadPool = transport.getThreadPool();
         final Transport.RequestHandlers requestHandlers = transport.getRequestHandlers();
         this.pipeline = new InboundPipeline(
-            transport.getVersion(),
             transport.getStatsTracker(),
             recycler,
             threadPool::relativeTimeInMillis,
diff --git a/server/src/main/java/org/elasticsearch/transport/Header.java b/server/src/main/java/org/elasticsearch/transport/Header.java
@@ -49,10 +49,6 @@ long getRequestId() {
         return requestId;
     }
 
-    byte getStatus() {
-        return status;
-    }
-
     boolean isRequest() {
         return TransportStatus.isRequest(status);
     }
diff --git a/server/src/main/java/org/elasticsearch/transport/InboundAggregator.java b/server/src/main/java/org/elasticsearch/transport/InboundAggregator.java
@@ -40,7 +40,7 @@ public InboundAggregator(
         Function<String, RequestHandlerRegistry<TransportRequest>> registryFunction,
         boolean ignoreDeserializationErrors
     ) {
-        this(circuitBreaker, (Predicate<String>) actionName -> {
+        this(circuitBreaker, actionName -> {
             final RequestHandlerRegistry<TransportRequest> reg = registryFunction.apply(actionName);
             if (reg == null) {
                 assert ignoreDeserializationErrors : actionName;
diff --git a/server/src/main/java/org/elasticsearch/transport/InboundDecoder.java b/server/src/main/java/org/elasticsearch/transport/InboundDecoder.java
@@ -25,16 +25,14 @@ public class InboundDecoder implements Releasable {
     static final Object PING = new Object();
     static final Object END_CONTENT = new Object();
 
-    private final TransportVersion version;
     private final Recycler<BytesRef> recycler;
     private TransportDecompressor decompressor;
     private int totalNetworkSize = -1;
     private int bytesConsumed = 0;
     private boolean isCompressed = false;
     private boolean isClosed = false;
 
-    public InboundDecoder(TransportVersion version, Recycler<BytesRef> recycler) {
-        this.version = version;
+    public InboundDecoder(Recycler<BytesRef> recycler) {
         this.recycler = recycler;
     }
 
diff --git a/server/src/main/java/org/elasticsearch/transport/InboundPipeline.java b/server/src/main/java/org/elasticsearch/transport/InboundPipeline.java
@@ -9,7 +9,6 @@
 package org.elasticsearch.transport;
 
 import org.apache.lucene.util.BytesRef;
-import org.elasticsearch.TransportVersion;
 import org.elasticsearch.common.breaker.CircuitBreaker;
 import org.elasticsearch.common.bytes.CompositeBytesReference;
 import org.elasticsearch.common.bytes.ReleasableBytesReference;
@@ -40,7 +39,6 @@ public class InboundPipeline implements Releasable {
     private boolean isClosed = false;
 
     public InboundPipeline(
-        TransportVersion version,
         StatsTracker statsTracker,
         Recycler<BytesRef> recycler,
         LongSupplier relativeTimeInMillis,
@@ -52,7 +50,7 @@ public InboundPipeline(
         this(
             statsTracker,
             relativeTimeInMillis,
-            new InboundDecoder(version, recycler),
+            new InboundDecoder(recycler),
             new InboundAggregator(circuitBreaker, registryFunction, ignoreDeserializationErrors),
             messageHandler
         );
diff --git a/server/src/main/java/org/elasticsearch/transport/NetworkMessage.java b/server/src/main/java/org/elasticsearch/transport/NetworkMessage.java
@@ -41,26 +41,10 @@ public abstract class NetworkMessage {
         }
     }
 
-    public TransportVersion getVersion() {
-        return version;
-    }
-
-    public long getRequestId() {
-        return requestId;
-    }
-
     boolean isCompress() {
         return TransportStatus.isCompress(status);
     }
 
-    boolean isResponse() {
-        return TransportStatus.isRequest(status) == false;
-    }
-
-    boolean isRequest() {
-        return TransportStatus.isRequest(status);
-    }
-
     boolean isHandshake() {
         return TransportStatus.isHandshake(status);
     }
diff --git a/server/src/main/java/org/elasticsearch/transport/ProxyConnectionStrategy.java b/server/src/main/java/org/elasticsearch/transport/ProxyConnectionStrategy.java
@@ -397,22 +397,6 @@ public String modeName() {
             return "proxy";
         }
 
-        public String getAddress() {
-            return address;
-        }
-
-        public String getServerName() {
-            return serverName;
-        }
-
-        public int getMaxSocketConnections() {
-            return maxSocketConnections;
-        }
-
-        public int getNumSocketsConnected() {
-            return numSocketsConnected;
-        }
-
         @Override
         public RemoteConnectionStrategy.ConnectionStrategy modeType() {
             return RemoteConnectionStrategy.ConnectionStrategy.PROXY;
diff --git a/server/src/main/java/org/elasticsearch/transport/RemoteClusterPortSettings.java b/server/src/main/java/org/elasticsearch/transport/RemoteClusterPortSettings.java
@@ -133,7 +133,7 @@ public class RemoteClusterPortSettings {
 
     static void validateRemoteAccessSettings(Settings settings) {
         if (REMOTE_CLUSTER_SERVER_ENABLED.get(settings)
-            && settings.getGroups("transport.profiles.", true).keySet().contains(REMOTE_CLUSTER_PROFILE)) {
+            && settings.getGroups("transport.profiles.", true).containsKey(REMOTE_CLUSTER_PROFILE)) {
             throw new IllegalArgumentException(
                 "Remote Access settings should not be configured using the ["
                     + REMOTE_CLUSTER_PROFILE
diff --git a/server/src/main/java/org/elasticsearch/transport/RemoteConnectionInfo.java b/server/src/main/java/org/elasticsearch/transport/RemoteConnectionInfo.java
@@ -85,14 +85,6 @@ public ModeInfo getModeInfo() {
         return modeInfo;
     }
 
-    public TimeValue getInitialConnectionTimeout() {
-        return initialConnectionTimeout;
-    }
-
-    public boolean isSkipUnavailable() {
-        return skipUnavailable;
-    }
-
     public boolean hasClusterCredentials() {
         return hasClusterCredentials;
     }
diff --git a/server/src/main/java/org/elasticsearch/transport/SniffConnectionStrategy.java b/server/src/main/java/org/elasticsearch/transport/SniffConnectionStrategy.java
@@ -621,18 +621,6 @@ public String modeName() {
             return "sniff";
         }
 
-        public List<String> getSeedNodes() {
-            return seedNodes;
-        }
-
-        public int getMaxConnectionsPerCluster() {
-            return maxConnectionsPerCluster;
-        }
-
-        public int getNumNodesConnected() {
-            return numNodesConnected;
-        }
-
         @Override
         public RemoteConnectionStrategy.ConnectionStrategy modeType() {
             return RemoteConnectionStrategy.ConnectionStrategy.SNIFF;
diff --git a/server/src/main/java/org/elasticsearch/transport/TransportService.java b/server/src/main/java/org/elasticsearch/transport/TransportService.java
@@ -750,21 +750,38 @@ public final <T extends TransportResponse> void sendRequest(
         final TransportRequestOptions options,
         TransportResponseHandler<T> handler
     ) {
-        final Transport.Connection connection;
+        final Transport.Connection connection = getConnectionOrFail(node, action, handler);
+        if (connection != null) {
+            sendRequest(connection, action, request, options, handler);
+        }
+    }
+
+    /**
+     * Get connection to the given {@code node} if possible and fail {@code handler} otherwise.
+     *
+     * @return connection if one could be found to {@code node} or null otherwise
+     */
+    @Nullable
+    private Transport.Connection getConnectionOrFail(DiscoveryNode node, String action, TransportResponseHandler<?> handler) {
         try {
-            connection = getConnection(node);
+            var connection = getConnection(node);
+            if (connection == null) {
+                final var ex = new NodeNotConnectedException(node, "Node not connected");
+                assert false : ex;
+                throw ex;
+            }
+            return connection;
         } catch (TransportException transportException) {
             // should only be a NodeNotConnectedException in practice, but handle all cases anyway to be sure
             assert transportException instanceof NodeNotConnectedException : transportException;
             handleSendRequestException(handler, transportException);
-            return;
+            return null;
         } catch (Exception exception) {
             // shouldn't happen in practice, but handle it anyway to be sure
             assert false : exception;
             handleSendRequestException(handler, new SendRequestTransportException(node, action, exception));
-            return;
+            return null;
         }
-        sendRequest(connection, action, request, options, handler);
     }
 
     /**
@@ -833,10 +850,7 @@ public final <T extends TransportResponse> void sendRequest(
         }
     }
 
-    private static <T extends TransportResponse> void handleSendRequestException(
-        TransportResponseHandler<T> handler,
-        TransportException transportException
-    ) {
+    private static void handleSendRequestException(TransportResponseHandler<?> handler, TransportException transportException) {
         try {
             handler.handleException(transportException);
         } catch (Exception innerException) {
@@ -867,21 +881,10 @@ public final <T extends TransportResponse> void sendChildRequest(
         final TransportRequestOptions options,
         final TransportResponseHandler<T> handler
     ) {
-        final Transport.Connection connection;
-        try {
-            connection = getConnection(node);
-        } catch (TransportException transportException) {
-            // should only be a NodeNotConnectedException in practice, but handle all cases anyway to be sure
-            assert transportException instanceof NodeNotConnectedException : transportException;
-            handleSendRequestException(handler, transportException);
-            return;
-        } catch (Exception exception) {
-            // shouldn't happen in practice, but handle it anyway to be sure
-            assert false : exception;
-            handleSendRequestException(handler, new SendRequestTransportException(node, action, exception));
-            return;
+        final Transport.Connection connection = getConnectionOrFail(node, action, handler);
+        if (connection != null) {
+            sendChildRequest(connection, action, request, parentTask, options, handler);
         }
-        sendChildRequest(connection, action, request, parentTask, options, handler);
     }
 
     public <T extends TransportResponse> void sendChildRequest(
diff --git a/server/src/main/java/org/elasticsearch/transport/TransportStats.java b/server/src/main/java/org/elasticsearch/transport/TransportStats.java
@@ -121,14 +121,6 @@ public long getServerOpen() {
         return serverOpen();
     }
 
-    public long totalOutboundConnections() {
-        return this.totalOutboundConnections;
-    }
-
-    public long getTotalOutboundConnections() {
-        return totalOutboundConnections();
-    }
-
     public long rxCount() {
         return rxCount;
     }
@@ -186,31 +178,29 @@ private boolean assertHistogramsConsistent() {
 
     @Override
     public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params outerParams) {
-        return Iterators.<ToXContent>concat(
-
-            Iterators.single((builder, params) -> {
-                builder.startObject(Fields.TRANSPORT);
-                builder.field(Fields.SERVER_OPEN, serverOpen);
-                builder.field(Fields.TOTAL_OUTBOUND_CONNECTIONS, totalOutboundConnections);
-                builder.field(Fields.RX_COUNT, rxCount);
-                builder.humanReadableField(Fields.RX_SIZE_IN_BYTES, Fields.RX_SIZE, ByteSizeValue.ofBytes(rxSize));
-                builder.field(Fields.TX_COUNT, txCount);
-                builder.humanReadableField(Fields.TX_SIZE_IN_BYTES, Fields.TX_SIZE, ByteSizeValue.ofBytes(txSize));
-                if (inboundHandlingTimeBucketFrequencies.length > 0) {
-                    histogramToXContent(builder, inboundHandlingTimeBucketFrequencies, Fields.INBOUND_HANDLING_TIME_HISTOGRAM);
-                    histogramToXContent(builder, outboundHandlingTimeBucketFrequencies, Fields.OUTBOUND_HANDLING_TIME_HISTOGRAM);
-                } else {
-                    // Stats came from before v8.1
-                    assert Version.CURRENT.major == Version.V_7_0_0.major + 1;
-                }
-                if (transportActionStats.isEmpty() == false) {
-                    builder.startObject(Fields.ACTIONS);
-                } else {
-                    // Stats came from before v8.8
-                    assert Version.CURRENT.major == Version.V_7_0_0.major + 1;
-                }
-                return builder;
-            }),
+        return Iterators.concat(Iterators.single((builder, params) -> {
+            builder.startObject(Fields.TRANSPORT);
+            builder.field(Fields.SERVER_OPEN, serverOpen);
+            builder.field(Fields.TOTAL_OUTBOUND_CONNECTIONS, totalOutboundConnections);
+            builder.field(Fields.RX_COUNT, rxCount);
+            builder.humanReadableField(Fields.RX_SIZE_IN_BYTES, Fields.RX_SIZE, ByteSizeValue.ofBytes(rxSize));
+            builder.field(Fields.TX_COUNT, txCount);
+            builder.humanReadableField(Fields.TX_SIZE_IN_BYTES, Fields.TX_SIZE, ByteSizeValue.ofBytes(txSize));
+            if (inboundHandlingTimeBucketFrequencies.length > 0) {
+                histogramToXContent(builder, inboundHandlingTimeBucketFrequencies, Fields.INBOUND_HANDLING_TIME_HISTOGRAM);
+                histogramToXContent(builder, outboundHandlingTimeBucketFrequencies, Fields.OUTBOUND_HANDLING_TIME_HISTOGRAM);
+            } else {
+                // Stats came from before v8.1
+                assert Version.CURRENT.major == Version.V_7_0_0.major + 1;
+            }
+            if (transportActionStats.isEmpty() == false) {
+                builder.startObject(Fields.ACTIONS);
+            } else {
+                // Stats came from before v8.8
+                assert Version.CURRENT.major == Version.V_7_0_0.major + 1;
+            }
+            return builder;
+        }),
 
             Iterators.map(transportActionStats.entrySet().iterator(), entry -> (builder, params) -> {
                 builder.field(entry.getKey());
diff --git a/server/src/test/java/org/elasticsearch/transport/InboundDecoderTests.java b/server/src/test/java/org/elasticsearch/transport/InboundDecoderTests.java
@@ -77,7 +77,7 @@ public void testDecode() throws IOException {
             );
             final BytesReference messageBytes = totalBytes.slice(totalHeaderSize, totalBytes.length() - totalHeaderSize);
 
-            InboundDecoder decoder = new InboundDecoder(TransportVersion.current(), recycler);
+            InboundDecoder decoder = new InboundDecoder(recycler);
             final ArrayList<Object> fragments = new ArrayList<>();
             final ReleasableBytesReference releasable1 = ReleasableBytesReference.wrap(totalBytes);
             int bytesConsumed = decoder.decode(releasable1, fragments::add);
@@ -141,7 +141,7 @@ public void testDecodePreHeaderSizeVariableInt() throws IOException {
             final BytesReference totalBytes = message.serialize(os);
             int partialHeaderSize = TcpHeader.headerSize(preHeaderVariableInt);
 
-            InboundDecoder decoder = new InboundDecoder(TransportVersion.current(), recycler);
+            InboundDecoder decoder = new InboundDecoder(recycler);
             final ArrayList<Object> fragments = new ArrayList<>();
             final ReleasableBytesReference releasable1 = ReleasableBytesReference.wrap(totalBytes);
             int bytesConsumed = decoder.decode(releasable1, fragments::add);
@@ -198,7 +198,7 @@ public void testDecodeHandshakeCompatibility() throws IOException {
             final BytesReference bytes = message.serialize(os);
             int totalHeaderSize = TcpHeader.headerSize(handshakeCompat);
 
-            InboundDecoder decoder = new InboundDecoder(TransportVersion.current(), recycler);
+            InboundDecoder decoder = new InboundDecoder(recycler);
             final ArrayList<Object> fragments = new ArrayList<>();
             final ReleasableBytesReference releasable1 = ReleasableBytesReference.wrap(bytes);
             int bytesConsumed = decoder.decode(releasable1, fragments::add);
@@ -257,7 +257,7 @@ public void testCompressedDecode() throws IOException {
                 TcpHeader.VARIABLE_HEADER_SIZE_POSITION
             );
 
-            InboundDecoder decoder = new InboundDecoder(TransportVersion.current(), recycler);
+            InboundDecoder decoder = new InboundDecoder(recycler);
             final ArrayList<Object> fragments = new ArrayList<>();
             final ReleasableBytesReference releasable1 = ReleasableBytesReference.wrap(totalBytes);
             int bytesConsumed = decoder.decode(releasable1, fragments::add);
@@ -321,7 +321,7 @@ public void testCompressedDecodeHandshakeCompatibility() throws IOException {
             final BytesReference bytes = message.serialize(os);
             int totalHeaderSize = TcpHeader.headerSize(handshakeCompat);
 
-            InboundDecoder decoder = new InboundDecoder(TransportVersion.current(), recycler);
+            InboundDecoder decoder = new InboundDecoder(recycler);
             final ArrayList<Object> fragments = new ArrayList<>();
             final ReleasableBytesReference releasable1 = ReleasableBytesReference.wrap(bytes);
             int bytesConsumed = decoder.decode(releasable1, fragments::add);
@@ -358,7 +358,7 @@ public void testVersionIncompatibilityDecodeException() throws IOException {
         try (RecyclerBytesStreamOutput os = new RecyclerBytesStreamOutput(recycler)) {
             final BytesReference bytes = message.serialize(os);
 
-            InboundDecoder decoder = new InboundDecoder(TransportVersion.current(), recycler);
+            InboundDecoder decoder = new InboundDecoder(recycler);
             final ArrayList<Object> fragments = new ArrayList<>();
             try (ReleasableBytesReference r = ReleasableBytesReference.wrap(bytes)) {
                 releasable1 = r;
diff --git a/server/src/test/java/org/elasticsearch/transport/InboundPipelineTests.java b/server/src/test/java/org/elasticsearch/transport/InboundPipelineTests.java
@@ -78,7 +78,7 @@ public void testPipelineHandling() throws IOException {
 
         final StatsTracker statsTracker = new StatsTracker();
         final LongSupplier millisSupplier = () -> TimeValue.nsecToMSec(System.nanoTime());
-        final InboundDecoder decoder = new InboundDecoder(TransportVersion.current(), recycler);
+        final InboundDecoder decoder = new InboundDecoder(recycler);
         final String breakThisAction = "break_this_action";
         final String actionName = "actionName";
         final Predicate<String> canTripBreaker = breakThisAction::equals;
@@ -208,7 +208,7 @@ public void testDecodeExceptionIsPropagated() throws IOException {
         BiConsumer<TcpChannel, InboundMessage> messageHandler = (c, m) -> {};
         final StatsTracker statsTracker = new StatsTracker();
         final LongSupplier millisSupplier = () -> TimeValue.nsecToMSec(System.nanoTime());
-        final InboundDecoder decoder = new InboundDecoder(TransportVersion.current(), recycler);
+        final InboundDecoder decoder = new InboundDecoder(recycler);
         final Supplier<CircuitBreaker> breaker = () -> new NoopCircuitBreaker("test");
         final InboundAggregator aggregator = new InboundAggregator(breaker, (Predicate<String>) action -> true);
         final InboundPipeline pipeline = new InboundPipeline(statsTracker, millisSupplier, decoder, aggregator, messageHandler);
@@ -253,7 +253,7 @@ public void testEnsureBodyIsNotPrematurelyReleased() throws IOException {
         BiConsumer<TcpChannel, InboundMessage> messageHandler = (c, m) -> {};
         final StatsTracker statsTracker = new StatsTracker();
         final LongSupplier millisSupplier = () -> TimeValue.nsecToMSec(System.nanoTime());
-        final InboundDecoder decoder = new InboundDecoder(TransportVersion.current(), recycler);
+        final InboundDecoder decoder = new InboundDecoder(recycler);
         final Supplier<CircuitBreaker> breaker = () -> new NoopCircuitBreaker("test");
         final InboundAggregator aggregator = new InboundAggregator(breaker, (Predicate<String>) action -> true);
         final InboundPipeline pipeline = new InboundPipeline(statsTracker, millisSupplier, decoder, aggregator, messageHandler);
diff --git a/server/src/test/java/org/elasticsearch/transport/OutboundHandlerTests.java b/server/src/test/java/org/elasticsearch/transport/OutboundHandlerTests.java

Original file line number	Diff line number	Diff line change
`@@ -49,10 +49,6 @@ long getRequestId() {`
`49`	`49`	`return requestId;`
`50`	`50`	`}`
`51`	`51`
`52`		`- byte getStatus() {`
`53`		`- return status;`
`54`		`- }`
`55`		`-`
`56`	`52`	`boolean isRequest() {`
`57`	`53`	`return TransportStatus.isRequest(status);`
`58`	`54`	`}`
Original file line number	Diff line number	Diff line change
`@@ -85,14 +85,6 @@ public ModeInfo getModeInfo() {`
`85`	`85`	`return modeInfo;`
`86`	`86`	`}`
`87`	`87`
`88`		`- public TimeValue getInitialConnectionTimeout() {`
`89`		`- return initialConnectionTimeout;`
`90`		`- }`
`91`		`-`
`92`		`- public boolean isSkipUnavailable() {`
`93`		`- return skipUnavailable;`
`94`		`- }`
`95`		`-`
`96`	`88`	`public boolean hasClusterCredentials() {`
`97`	`89`	`return hasClusterCredentials;`
`98`	`90`	`}`