Enhanced logging when transport is misconfigured to talk to HTTP port

server/src/main/java/org/elasticsearch/ElasticsearchException.java

@@ -961,8 +961,8 @@ private enum ElasticsearchExceptionHandle {
         RESOURCE_ALREADY_EXISTS_EXCEPTION(ResourceAlreadyExistsException.class,
             ResourceAlreadyExistsException::new, 123, UNKNOWN_VERSION_ADDED),
         // 124 used to be Script.ScriptParseException
-        HTTP_ON_TRANSPORT_EXCEPTION(TcpTransport.HttpOnTransportException.class,
-                TcpTransport.HttpOnTransportException::new, 125, UNKNOWN_VERSION_ADDED),
+        HTTP_REQUEST_ON_TRANSPORT_EXCEPTION(TcpTransport.HttpRequestOnTransportException.class,
+                TcpTransport.HttpRequestOnTransportException::new, 125, UNKNOWN_VERSION_ADDED),
         MAPPER_PARSING_EXCEPTION(org.elasticsearch.index.mapper.MapperParsingException.class,
                 org.elasticsearch.index.mapper.MapperParsingException::new, 126, UNKNOWN_VERSION_ADDED),
         SEARCH_CONTEXT_EXCEPTION(org.elasticsearch.search.SearchContextException.class,

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

@@ -601,7 +601,7 @@ public void onException(TcpChannel channel, Exception e) {
                 "cancelled key exception caught on transport layer [{}], disconnecting from relevant node", channel), e);
             // close the channel as safe measure, which will cause a node to be disconnected if relevant
             CloseableChannel.closeChannel(channel);
-        } else if (e instanceof TcpTransport.HttpOnTransportException) {
+        } else if (e instanceof HttpRequestOnTransportException) {
             // in case we are able to return data, serialize the exception content and sent it back to the client
             if (channel.isOpen()) {
                 BytesArray message = new BytesArray(e.getMessage().getBytes(StandardCharsets.UTF_8));
@@ -674,7 +674,7 @@ public void inboundMessage(TcpChannel channel, BytesReference message) {
      * @param bytesReference the bytes available to consume
      * @return the number of bytes consumed
      * @throws StreamCorruptedException              if the message header format is not recognized
-     * @throws TcpTransport.HttpOnTransportException if the message header appears to be an HTTP message
+     * @throws HttpRequestOnTransportException       if the message header appears to be an HTTP message
      * @throws IllegalArgumentException              if the message length is greater that the maximum allowed frame size.
      *                                               This is dependent on the available memory.
      */
@@ -696,7 +696,7 @@ public int consumeNetworkReads(TcpChannel channel, BytesReference bytesReference
      * @param networkBytes the will be read
      * @return the message decoded
      * @throws StreamCorruptedException              if the message header format is not recognized
-     * @throws TcpTransport.HttpOnTransportException if the message header appears to be an HTTP message
+     * @throws HttpRequestOnTransportException       if the message header appears to be an HTTP message
      * @throws IllegalArgumentException              if the message length is greater that the maximum allowed frame size.
      *                                               This is dependent on the available memory.
      */
@@ -723,7 +723,7 @@ static BytesReference decodeFrame(BytesReference networkBytes) throws IOExceptio
      * @param networkBytes the will be read
      * @return the length of the message
      * @throws StreamCorruptedException              if the message header format is not recognized
-     * @throws TcpTransport.HttpOnTransportException if the message header appears to be an HTTP message
+     * @throws HttpRequestOnTransportException       if the message header appears to be an HTTP message
      * @throws IllegalArgumentException              if the message length is greater that the maximum allowed frame size.
      *                                               This is dependent on the available memory.
      */
@@ -737,8 +737,13 @@ public static int readMessageLength(BytesReference networkBytes) throws IOExcept
 
     private static int readHeaderBuffer(BytesReference headerBuffer) throws IOException {
         if (headerBuffer.get(0) != 'E' || headerBuffer.get(1) != 'S') {
-            if (appearsToBeHTTP(headerBuffer)) {
-                throw new TcpTransport.HttpOnTransportException("This is not an HTTP port");
+            if (appearsToBeHTTPRequest(headerBuffer)) {
+                throw new HttpRequestOnTransportException("This is not an HTTP port");
+            }
+
+            if (appearsToBeHTTPResponse(headerBuffer)) {
+                throw new StreamCorruptedException("received HTTP response on transport port, ensure that transport port (not " +
+                        "HTTP port) of a remote node is specified in the configuration");
             }
 
             String firstBytes = "("
@@ -772,7 +777,7 @@ private static int readHeaderBuffer(BytesReference headerBuffer) throws IOExcept
         return messageLength;
     }
 
-    private static boolean appearsToBeHTTP(BytesReference headerBuffer) {
+    private static boolean appearsToBeHTTPRequest(BytesReference headerBuffer) {
         return bufferStartsWith(headerBuffer, "GET") ||
             bufferStartsWith(headerBuffer, "POST") ||
             bufferStartsWith(headerBuffer, "PUT") ||
@@ -784,6 +789,10 @@ private static boolean appearsToBeHTTP(BytesReference headerBuffer) {
             bufferStartsWith(headerBuffer, "TRACE");
     }
 
+    private static boolean appearsToBeHTTPResponse(BytesReference headerBuffer) {
+        return bufferStartsWith(headerBuffer, "HTTP");
+    }
+
     private static boolean appearsToBeTLS(BytesReference headerBuffer) {
         return headerBuffer.get(0) == 0x16 && headerBuffer.get(1) == 0x03;
     }
@@ -802,9 +811,9 @@ private static boolean bufferStartsWith(BytesReference buffer, String method) {
      * A helper exception to mark an incoming connection as potentially being HTTP
      * so an appropriate error code can be returned
      */
-    public static class HttpOnTransportException extends ElasticsearchException {
+    public static class HttpRequestOnTransportException extends ElasticsearchException {
 
-        private HttpOnTransportException(String msg) {
+        private HttpRequestOnTransportException(String msg) {
             super(msg);
         }
 
@@ -813,7 +822,7 @@ public RestStatus status() {
             return RestStatus.BAD_REQUEST;
         }
 
-        public HttpOnTransportException(StreamInput in) throws IOException {
+        public HttpRequestOnTransportException(StreamInput in) throws IOException {
             super(in);
         }
     }

server/src/test/java/org/elasticsearch/ExceptionSerializationTests.java

@@ -787,7 +787,7 @@ public void testIds() {
         ids.put(122, null);
         ids.put(123, org.elasticsearch.ResourceAlreadyExistsException.class);
         ids.put(124, null);
-        ids.put(125, TcpTransport.HttpOnTransportException.class);
+        ids.put(125, TcpTransport.HttpRequestOnTransportException.class);
         ids.put(126, org.elasticsearch.index.mapper.MapperParsingException.class);
         ids.put(127, org.elasticsearch.search.SearchContextException.class);
         ids.put(128, org.elasticsearch.search.builder.SearchSourceBuilderException.class);

server/src/test/java/org/elasticsearch/transport/TcpTransportTests.java

@@ -288,6 +288,28 @@ public void testInvalidHeader() throws IOException {
         }
     }
 
+    public void testHTTPRequest() throws IOException {
+        String[] httpHeaders = {"GET", "POST", "PUT", "HEAD", "DELETE", "OPTIONS", "PATCH", "TRACE"};
+
+        for (String httpHeader : httpHeaders) {
+            BytesStreamOutput streamOutput = new BytesStreamOutput(1 << 14);
+
+            for (char c : httpHeader.toCharArray()) {
+                streamOutput.write((byte) c);
+            }
+            streamOutput.write(new byte[6]);
+
+            try {
+                BytesReference bytes = streamOutput.bytes();
+                TcpTransport.decodeFrame(bytes);
+                fail("Expected exception");
+            } catch (Exception ex) {
+                assertThat(ex, instanceOf(TcpTransport.HttpRequestOnTransportException.class));
+                assertEquals("This is not an HTTP port", ex.getMessage());
+            }
+        }
+    }
+
     public void testTLSHeader() throws IOException {
         BytesStreamOutput streamOutput = new BytesStreamOutput(1 << 14);
 
@@ -314,25 +336,22 @@ public void testTLSHeader() throws IOException {
         }
     }
 
-    public void testHTTPHeader() throws IOException {
-        String[] httpHeaders = {"GET", "POST", "PUT", "HEAD", "DELETE", "OPTIONS", "PATCH", "TRACE"};
-
-        for (String httpHeader : httpHeaders) {
-            BytesStreamOutput streamOutput = new BytesStreamOutput(1 << 14);
-
-            for (char c : httpHeader.toCharArray()) {
-                streamOutput.write((byte) c);
-            }
-            streamOutput.write(new byte[6]);
+    public void testHTTPResponse() throws IOException {
+        BytesStreamOutput streamOutput = new BytesStreamOutput(1 << 14);
+        streamOutput.write('H');
+        streamOutput.write('T');
+        streamOutput.write('T');
+        streamOutput.write('P');
+        streamOutput.write(randomByte());
+        streamOutput.write(randomByte());
 
-            try {
-                BytesReference bytes = streamOutput.bytes();
-                TcpTransport.decodeFrame(bytes);
-                fail("Expected exception");
-            } catch (Exception ex) {
-                assertThat(ex, instanceOf(TcpTransport.HttpOnTransportException.class));
-                assertEquals("This is not an HTTP port", ex.getMessage());
-            }
+        try {
+            TcpTransport.decodeFrame(streamOutput.bytes());
+            fail("Expected exception");
+        } catch (Exception ex) {
+            assertThat(ex, instanceOf(StreamCorruptedException.class));
+            assertEquals("received HTTP response on transport port, ensure that transport port " +
+                    "(not HTTP port) of a remote node is specified in the configuration", ex.getMessage());
         }
     }
 }