Use the Upgrade flow by default

Signed-off-by: Andriy Redko <[email protected]>

client/rest/licenses/httpclient5-5.4.1.jar.sha1

@@ -1 +1 @@
-ce913081e592ee8eeee35c4e577d7dce13cba7a4
+ce913081e592ee8eeee35c4e577d7dce13cba7a4

client/rest/licenses/httpcore5-5.3.1.jar.sha1

@@ -1 +1 @@
-eaf64237945d7d0f301d48420e8bdb7f565a7b0e
+eaf64237945d7d0f301d48420e8bdb7f565a7b0e

client/rest/licenses/httpcore5-h2-5.3.1.jar.sha1

@@ -1 +1 @@
-760c34db3ba41b0ffa07e956bc308d3a12356915
+760c34db3ba41b0ffa07e956bc308d3a12356915

client/rest/licenses/httpcore5-reactive-5.3.1.jar.sha1

@@ -1 +1 @@
-c4c0c3c7bbcb0db54aa7ddd39e34a835428c99c0
+c4c0c3c7bbcb0db54aa7ddd39e34a835428c99c0

client/rest/src/main/java/org/opensearch/client/RestClientBuilder.java

@@ -308,15 +308,10 @@ private CloseableHttpAsyncClient createHttpClient() {
         // default timeouts are all infinite
         RequestConfig.Builder requestConfigBuilder = RequestConfig.custom()
             .setConnectTimeout(Timeout.ofMilliseconds(DEFAULT_CONNECT_TIMEOUT_MILLIS))
-            .setResponseTimeout(Timeout.ofMilliseconds(DEFAULT_RESPONSE_TIMEOUT_MILLIS))
-            .setProtocolUpgradeEnabled(false);
+            .setResponseTimeout(Timeout.ofMilliseconds(DEFAULT_RESPONSE_TIMEOUT_MILLIS));
         if (requestConfigCallback != null) {
             requestConfigBuilder = requestConfigCallback.customizeRequestConfig(requestConfigBuilder);
         }
-        RequestConfig requestConfig = requestConfigBuilder.build();
-        if (requestConfig.isProtocolUpgradeEnabled()) {
-            throw new IllegalArgumentException("protocol upgrade is not supported");
-        }
 
         try {
             final TlsStrategy tlsStrategy = ClientTlsStrategyBuilder.create()
@@ -337,7 +332,7 @@ public TlsDetails create(final SSLEngine sslEngine) {
                 .build();
 
             HttpAsyncClientBuilder httpClientBuilder = HttpAsyncClientBuilder.create()
-                .setDefaultRequestConfig(requestConfig)
+                .setDefaultRequestConfig(requestConfigBuilder.build())
                 .setConnectionManager(connectionManager)
                 .setTargetAuthenticationStrategy(DefaultAuthenticationStrategy.INSTANCE)
                 .disableAutomaticRetries();

client/rest/src/test/java/org/opensearch/client/RestClientSingleHostIntegTests.java

@@ -382,6 +382,10 @@ public void testHeaders() throws Exception {
             if (method.equals("HEAD") == false) {
                 standardHeaders.add("Content-length");
             }
+            if (method.equals("HEAD") == true || method.equals("GET") == true || method.equals("OPTIONS") == true) {
+                standardHeaders.add("Upgrade");
+            }
+
             final Header[] requestHeaders = RestClientTestUtil.randomHeaders(getRandom(), "Header");
             final int statusCode = randomStatusCode(getRandom());
             Request request = new Request(method, "/" + statusCode);
@@ -400,11 +404,15 @@ public void testHeaders() throws Exception {
             assertEquals(method, esResponse.getRequestLine().getMethod());
             assertEquals(statusCode, esResponse.getStatusLine().getStatusCode());
             assertEquals(pathPrefix + "/" + statusCode, esResponse.getRequestLine().getUri());
+
             assertHeaders(defaultHeaders, requestHeaders, esResponse.getHeaders(), standardHeaders);
+            final Set<String> removedHeaders = new HashSet<>();
             for (final Header responseHeader : esResponse.getHeaders()) {
                 String name = responseHeader.getName();
-                if (name.startsWith("Header") == false) {
+                // Some headers could be returned multiple times in response, like Connection fe.
+                if (name.startsWith("Header") == false && removedHeaders.contains(name) == false) {
                     assertTrue("unknown header was returned " + name, standardHeaders.remove(name));
+                    removedHeaders.add(name);
                 }
             }
             assertTrue("some expected standard headers weren't returned: " + standardHeaders, standardHeaders.isEmpty());

client/sniffer/licenses/httpclient5-5.4.1.jar.sha1

@@ -1 +1 @@
-ce913081e592ee8eeee35c4e577d7dce13cba7a4
+ce913081e592ee8eeee35c4e577d7dce13cba7a4

client/sniffer/licenses/httpcore5-5.3.1.jar.sha1

@@ -1 +1 @@
-eaf64237945d7d0f301d48420e8bdb7f565a7b0e
+eaf64237945d7d0f301d48420e8bdb7f565a7b0e

plugins/transport-reactor-netty4/src/javaRestTest/java/org/opensearch/rest/ReactorNetty4BadRequestIT.java

@@ -124,4 +124,13 @@ public void testUnsupportedContentType() throws IOException {
         final String error = objectPath.evaluate("error");
         assertThat(error, equalTo("Content-Type header [] is not supported"));
     }
+
+    @Override
+    protected final Settings restClientSettings() {
+        return Settings.builder()
+            .put(super.restClientSettings())
+            // See please https://github.com/reactor/reactor-netty/issues/3538
+            .put(OpenSearchRestTestCase.CLIENT_PROTOCOL_UPGRADE_ENABLED, false)
+            .build();
+    }
 }

plugins/transport-reactor-netty4/src/javaRestTest/java/org/opensearch/rest/ReactorNetty4HeadBodyIsEmptyIT.java

@@ -34,6 +34,7 @@
 
 import org.opensearch.client.Request;
 import org.opensearch.client.Response;
+import org.opensearch.common.settings.Settings;
 import org.opensearch.core.xcontent.XContentBuilder;
 import org.opensearch.test.rest.OpenSearchRestTestCase;
 import org.hamcrest.Matcher;
@@ -201,4 +202,12 @@ private void headTestCase(
         assertNull("HEAD requests shouldn't have a response body but " + url + " did", response.getEntity());
     }
 
+    @Override
+    protected final Settings restClientSettings() {
+        return Settings.builder()
+            .put(super.restClientSettings())
+            // See please https://github.com/reactor/reactor-netty/issues/3538
+            .put(OpenSearchRestTestCase.CLIENT_PROTOCOL_UPGRADE_ENABLED, false)
+            .build();
+    }
 }

plugins/transport-reactor-netty4/src/javaRestTest/java/org/opensearch/rest/ReactorNetty4StreamingIT.java

@@ -13,6 +13,7 @@
 import org.opensearch.client.ResponseException;
 import org.opensearch.client.StreamingRequest;
 import org.opensearch.client.StreamingResponse;
+import org.opensearch.common.settings.Settings;
 import org.opensearch.test.rest.OpenSearchRestTestCase;
 import org.opensearch.test.rest.yaml.ObjectPath;
 import org.junit.After;
@@ -350,4 +351,13 @@ public void testStreamingLargeDocumentThatExceedsChunkSize() throws IOException
         assertThat(streamingResponse.getStatusLine().getStatusCode(), equalTo(200));
         assertThat(streamingResponse.getWarnings(), empty());
     }
+
+    @Override
+    protected final Settings restClientSettings() {
+        return Settings.builder()
+            .put(super.restClientSettings())
+            // See please https://github.com/reactor/reactor-netty/issues/3538
+            .put(OpenSearchRestTestCase.CLIENT_PROTOCOL_UPGRADE_ENABLED, false)
+            .build();
+    }
 }

plugins/transport-reactor-netty4/src/javaRestTest/java/org/opensearch/rest/ReactorNetty4StreamingStressIT.java

@@ -13,6 +13,7 @@
 import org.opensearch.client.Response;
 import org.opensearch.client.StreamingRequest;
 import org.opensearch.client.StreamingResponse;
+import org.opensearch.common.settings.Settings;
 import org.opensearch.test.rest.OpenSearchRestTestCase;
 import org.junit.After;
 
@@ -79,4 +80,13 @@ public void testCloseClientStreamingRequest() throws Exception {
             .expectErrorMatches(t -> t instanceof InterruptedIOException || t instanceof ConnectionClosedException)
             .verify();
     }
+
+    @Override
+    protected final Settings restClientSettings() {
+        return Settings.builder()
+            .put(super.restClientSettings())
+            // See please https://github.com/reactor/reactor-netty/issues/3538
+            .put(OpenSearchRestTestCase.CLIENT_PROTOCOL_UPGRADE_ENABLED, false)
+            .build();
+    }
 }

test/framework/src/main/java/org/opensearch/bootstrap/BootstrapForTesting.java

@@ -301,15 +301,11 @@ static Set<URL> parseClassPathWithSymlinks() throws Exception {
     private static Set<String> getTrustedHosts() {
         //
         try {
-            List<String> hosts = Collections.list(NetworkInterface.getNetworkInterfaces())
+            return Collections.list(NetworkInterface.getNetworkInterfaces())
                 .stream()
                 .flatMap(iface -> Collections.list(iface.getInetAddresses()).stream())
                 .map(address -> NetworkAddress.format(address))
-                .collect(Collectors.toList());
-            // 0:0:0:0:0:0:0:1 is simplified to ::1, in it test, the incoming address can be 0:0:0:0:0:0:0:1,
-            // so we should add it to trusted hosts.
-            hosts.add("0:0:0:0:0:0:0:1");
-            return Collections.unmodifiableSet(new HashSet<>(hosts));
+                .collect(Collectors.toSet());
         } catch (final SocketException e) {
             return Collections.emptySet();
         }

test/framework/src/main/java/org/opensearch/test/rest/OpenSearchRestTestCase.java

@@ -135,6 +135,7 @@ public abstract class OpenSearchRestTestCase extends OpenSearchTestCase {
     public static final String TRUSTSTORE_PASSWORD = "truststore.password";
     public static final String CLIENT_SOCKET_TIMEOUT = "client.socket.timeout";
     public static final String CLIENT_PATH_PREFIX = "client.path.prefix";
+    public static final String CLIENT_PROTOCOL_UPGRADE_ENABLED = "client.protocol.upgrade.enabled";
 
     // This set will contain the warnings already asserted since we are eliminating logging duplicate warnings.
     // This ensures that no matter in what order the tests run, the warning is asserted once.
@@ -881,8 +882,10 @@ public TlsDetails create(final SSLEngine sslEngine) {
             socketTimeoutString == null ? "60s" : socketTimeoutString,
             CLIENT_SOCKET_TIMEOUT
         );
+        final boolean protocolUpgradeEnabled = settings.getAsBoolean(CLIENT_PROTOCOL_UPGRADE_ENABLED, true);
         builder.setRequestConfigCallback(
             conf -> conf.setResponseTimeout(Timeout.ofMilliseconds(Math.toIntExact(socketTimeout.getMillis())))
+                .setProtocolUpgradeEnabled(protocolUpgradeEnabled)
         );
         if (settings.hasValue(CLIENT_PATH_PREFIX)) {
             builder.setPathPrefix(settings.get(CLIENT_PATH_PREFIX));