Merge branch 'main' into st0

Signed-off-by: Sandesh Kumar <sandeshkr419@gmail.com>

CHANGELOG.md

@@ -19,12 +19,14 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Add _list/indices API as paginated alternate to _cat/indices ([#14718](https://github.com/opensearch-project/OpenSearch/pull/14718))
 - Add success and failure metrics for async shard fetch ([#15976](https://github.com/opensearch-project/OpenSearch/pull/15976))
 - Add new metric REMOTE_STORE to NodeStats API response ([#15611](https://github.com/opensearch-project/OpenSearch/pull/15611))
+- Add support to dynamically resize threadpools size. ([#16236](https://github.com/opensearch-project/OpenSearch/pull/16236))
 - [S3 Repository] Change default retry mechanism of s3 clients to Standard Mode ([#15978](https://github.com/opensearch-project/OpenSearch/pull/15978))
 - Add changes to block calls in cat shards, indices and segments based on dynamic limit settings ([#15986](https://github.com/opensearch-project/OpenSearch/pull/15986))
 - New `phone` & `phone-search` analyzer + tokenizer ([#15915](https://github.com/opensearch-project/OpenSearch/pull/15915))
 - Add _list/shards API as paginated alternate to _cat/shards ([#14641](https://github.com/opensearch-project/OpenSearch/pull/14641))
 - Latency and Memory allocation improvements to Multi Term Aggregation queries ([#14993](https://github.com/opensearch-project/OpenSearch/pull/14993))
 - Flat object field use IndexOrDocValuesQuery to optimize query ([#14383](https://github.com/opensearch-project/OpenSearch/issues/14383))
+- Add method to return dynamic SecureTransportParameters from SecureTransportSettingsProvider interface ([#16387](https://github.com/opensearch-project/OpenSearch/pull/16387)
 - [Star Tree - Search] Add support for metric aggregations with/without term query ([15289](https://github.com/opensearch-project/OpenSearch/pull/15289))
 
 ### Dependencies
@@ -82,6 +84,11 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Fix warnings from SLF4J on startup when repository-s3 is installed ([#16194](https://github.com/opensearch-project/OpenSearch/pull/16194))
 - Fix protobuf-java leak through client library dependencies ([#16254](https://github.com/opensearch-project/OpenSearch/pull/16254))
 - Fix multi-search with template doesn't return status code ([#16265](https://github.com/opensearch-project/OpenSearch/pull/16265))
+- [Streaming Indexing] Fix intermittent 'The bulk request must be terminated by a newline [\n]' failures [#16337](https://github.com/opensearch-project/OpenSearch/pull/16337))
+- Fix wrong default value when setting `index.number_of_routing_shards` to null on index creation ([#16331](https://github.com/opensearch-project/OpenSearch/pull/16331))
+- [Workload Management] Make query groups persistent across process restarts [#16370](https://github.com/opensearch-project/OpenSearch/pull/16370)
+- Fix inefficient Stream API call chains ending with count() ([#15386](https://github.com/opensearch-project/OpenSearch/pull/15386))
+- Fix array hashCode calculation in ResyncReplicationRequest ([#16378](https://github.com/opensearch-project/OpenSearch/pull/16378))
 
 ### Security
 

buildSrc/version.properties

@@ -34,7 +34,7 @@ netty             = 4.1.114.Final
 joda              = 2.12.7
 
 # project reactor
-reactor_netty     = 1.1.22
+reactor_netty     = 1.1.23
 reactor           = 3.5.20
 
 # client dependencies

distribution/src/config/jvm.options

@@ -85,3 +85,7 @@ ${error.file}
 
 # HDFS ForkJoinPool.common() support by SecurityManager
 -Djava.util.concurrent.ForkJoinPool.common.threadFactory=org.opensearch.secure_sm.SecuredForkJoinWorkerThreadFactory
+
+# See please https://bugs.openjdk.org/browse/JDK-8341127 (openjdk/jdk#21283)
+23:-XX:CompileCommand=dontinline,java/lang/invoke/MethodHandle.setAsTypeCache
+23:-XX:CompileCommand=dontinline,java/lang/invoke/MethodHandle.asTypeUncached

libs/core/src/main/java/org/opensearch/core/common/io/stream/Writeable.java

@@ -83,7 +83,7 @@ public static <R extends Reader<?>> void registerReader(final byte ordinal, fina
 
         public static void registerClassAlias(final Class<?> classInstance, final Class<?> classGeneric) {
             if (WRITER_CUSTOM_CLASS_MAP.putIfAbsent(classInstance, classGeneric) != null) {
-                throw new IllegalArgumentException("Streamable custom class already registered [" + classInstance.getClass() + "]");
+                throw new IllegalArgumentException("Streamable custom class already registered [" + classInstance.getName() + "]");
             }
         }
 
@@ -96,7 +96,7 @@ public static <W extends Writer<?>> W getWriter(final Class<?> clazz) {
         }
 
         /**
-         * Returns the ristered reader keyed by the unique ordinal
+         * Returns the registered reader keyed by the unique ordinal
          */
         @SuppressWarnings("unchecked")
         public static <R extends Reader<?>> R getReader(final byte b) {

libs/core/src/main/java/org/opensearch/core/xcontent/MediaTypeRegistry.java

@@ -57,6 +57,7 @@
 public final class MediaTypeRegistry {
     private static Map<String, MediaType> formatToMediaType = Map.of();
     private static Map<String, MediaType> typeWithSubtypeToMediaType = Map.of();
+    private static Map<String, MediaType> knownStringsToMediaType = Map.of();
 
     // Default mediaType singleton
     private static MediaType DEFAULT_MEDIA_TYPE;
@@ -84,6 +85,8 @@ private static void register(MediaType[] acceptedMediaTypes, Map<String, MediaTy
         // ensures the map is not overwritten:
         Map<String, MediaType> typeMap = new HashMap<>(typeWithSubtypeToMediaType);
         Map<String, MediaType> formatMap = new HashMap<>(formatToMediaType);
+        Map<String, MediaType> knownStringMap = new HashMap<>(knownStringsToMediaType);
+
         for (MediaType mediaType : acceptedMediaTypes) {
             if (formatMap.containsKey(mediaType.format())) {
                 throw new IllegalArgumentException("unable to register mediaType: [" + mediaType.format() + "]. Type already exists.");
@@ -107,13 +110,24 @@ private static void register(MediaType[] acceptedMediaTypes, Map<String, MediaTy
             MediaType mediaType = entry.getValue();
             typeMap.put(typeWithSubtype, mediaType);
             formatMap.putIfAbsent(mediaType.format(), mediaType); // ignore if the additional type mapping already exists
+            knownStringMap.put(mediaType.mediaType(), mediaType);
+            knownStringMap.put(mediaType.mediaTypeWithoutParameters(), mediaType);
         }
 
         formatToMediaType = Map.copyOf(formatMap);
         typeWithSubtypeToMediaType = Map.copyOf(typeMap);
+        knownStringsToMediaType = Map.copyOf(knownStringMap);
     }
 
     public static MediaType fromMediaType(String mediaType) {
+        if (mediaType == null) {
+            return null;
+        }
+        // Skip parsing if the string is an exact match for any known string value
+        final MediaType knownMediaType = knownStringsToMediaType.get(mediaType);
+        if (knownMediaType != null) {
+            return knownMediaType;
+        }
         ParsedMediaType parsedMediaType = parseMediaType(mediaType);
         return parsedMediaType != null ? parsedMediaType.getMediaType() : null;
     }

libs/core/src/test/java/org/opensearch/core/common/io/stream/WriteableTests.java

@@ -0,0 +1,30 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.core.common.io.stream;
+
+import org.opensearch.test.OpenSearchTestCase;
+import org.junit.Assert;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class WriteableTests extends OpenSearchTestCase {
+
+    public void testRegisterClassAlias() {
+        Writeable.WriteableRegistry.registerClassAlias(StringBuilder.class, AtomicInteger.class);
+        try {
+            Writeable.WriteableRegistry.registerClassAlias(StringBuilder.class, AtomicInteger.class);
+            Assert.fail("expected exception not thrown");
+        } catch (IllegalArgumentException illegalArgumentException) {
+            Assert.assertEquals(
+                "Streamable custom class already registered [java.lang.StringBuilder]",
+                illegalArgumentException.getMessage()
+            );
+        }
+    }
+}

modules/percolator/src/main/java/org/opensearch/percolator/QueryAnalyzer.java

@@ -510,19 +510,25 @@ static Result selectBestResult(Result result1, Result result2) {
     }
 
     private static int minTermLength(Set<QueryExtraction> extractions) {
-        // In case there are only range extractions, then we return Integer.MIN_VALUE,
-        // so that selectBestExtraction(...) we are likely to prefer the extractions that contains at least a single extraction
-        if (extractions.stream().filter(queryExtraction -> queryExtraction.term != null).count() == 0
-            && extractions.stream().filter(queryExtraction -> queryExtraction.range != null).count() > 0) {
-            return Integer.MIN_VALUE;
-        }
-
+        boolean hasTerm = false;
+        boolean hasRange = false;
         int min = Integer.MAX_VALUE;
+
         for (QueryExtraction qt : extractions) {
             if (qt.term != null) {
+                hasTerm = true;
                 min = Math.min(min, qt.bytes().length);
             }
+            if (qt.range != null) {
+                hasRange = true;
+            }
         }
+
+        // If there are no terms but there are ranges, return Integer.MIN_VALUE
+        if (!hasTerm && hasRange) {
+            return Integer.MIN_VALUE;
+        }
+
         return min;
     }
 

modules/transport-netty4/src/main/java/org/opensearch/transport/netty4/ssl/SecureNetty4Transport.java

@@ -142,9 +142,10 @@ public SSLServerChannelInitializer(String name) {
         protected void initChannel(Channel ch) throws Exception {
             super.initChannel(ch);
 
-            final boolean dualModeEnabled = NetworkModule.TRANSPORT_SSL_DUAL_MODE_ENABLED.get(settings);
+            final boolean dualModeEnabled = secureTransportSettingsProvider.parameters(settings)
+                .map(SecureTransportSettingsProvider.SecureTransportParameters::dualModeEnabled)
+                .orElse(false);
             if (dualModeEnabled) {
-                logger.info("SSL Dual mode enabled, using port unification handler");
                 final ChannelHandler portUnificationHandler = new DualModeSslHandler(
                     settings,
                     secureTransportSettingsProvider,
@@ -258,7 +259,9 @@ protected class SSLClientChannelInitializer extends Netty4Transport.ClientChanne
         public SSLClientChannelInitializer(DiscoveryNode node) {
             this.node = node;
 
-            final boolean dualModeEnabled = NetworkModule.TRANSPORT_SSL_DUAL_MODE_ENABLED.get(settings);
+            final boolean dualModeEnabled = secureTransportSettingsProvider.parameters(settings)
+                .map(SecureTransportSettingsProvider.SecureTransportParameters::dualModeEnabled)
+                .orElse(false);
             hostnameVerificationEnabled = NetworkModule.TRANSPORT_SSL_ENFORCE_HOSTNAME_VERIFICATION.get(settings);
             hostnameVerificationResolveHostName = NetworkModule.TRANSPORT_SSL_ENFORCE_HOSTNAME_VERIFICATION_RESOLVE_HOST_NAME.get(settings);
 

plugins/repository-azure/licenses/reactor-netty-core-1.1.23.jar.sha1

@@ -0,0 +1 @@
+a7059b0c18ab7aa0fa9e08b48cb6a20b15c11478

plugins/repository-azure/licenses/reactor-netty-http-1.1.23.jar.sha1

@@ -0,0 +1 @@
+94b294fa90aee2e88ad4337251e278aaac21362c

plugins/transport-reactor-netty4/licenses/reactor-netty-core-1.1.23.jar.sha1

@@ -0,0 +1 @@
+a7059b0c18ab7aa0fa9e08b48cb6a20b15c11478

plugins/transport-reactor-netty4/licenses/reactor-netty-http-1.1.23.jar.sha1

@@ -0,0 +1 @@
+94b294fa90aee2e88ad4337251e278aaac21362c

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

@@ -304,11 +304,36 @@ public void testStreamingLargeDocument() throws IOException {
             String.format(
                 Locale.getDefault(),
                 "{ \"index\": { \"_index\": \"test-streaming\", \"_id\": \"1\" } }\n{ \"name\": \"%s\"  }\n",
-                randomAlphaOfLength(5000)
+                randomAlphaOfLength(7000)
+            )
+        );
+
+        final StreamingRequest<ByteBuffer> streamingRequest = new StreamingRequest<>(
+            "POST",
+            "/_bulk/stream",
+            Flux.fromStream(stream).map(s -> ByteBuffer.wrap(s.getBytes(StandardCharsets.UTF_8)))
+        );
+
+        final StreamingResponse<ByteBuffer> streamingResponse = client().streamRequest(streamingRequest);
+
+        StepVerifier.create(Flux.from(streamingResponse.getBody()).map(b -> new String(b.array(), StandardCharsets.UTF_8)))
+            .expectNextMatches(s -> s.contains("\"result\":\"created\"") && s.contains("\"_id\":\"1\""))
+            .expectComplete()
+            .verify();
+
+        assertThat(streamingResponse.getStatusLine().getStatusCode(), equalTo(200));
+        assertThat(streamingResponse.getWarnings(), empty());
+    }
+
+    public void testStreamingLargeDocumentThatExceedsChunkSize() throws IOException {
+        final Stream<String> stream = Stream.of(
+            String.format(
+                Locale.getDefault(),
+                "{ \"index\": { \"_index\": \"test-streaming\", \"_id\": \"1\" } }\n{ \"name\": \"%s\"  }\n",
+                randomAlphaOfLength(9000) /* the default chunk size limit is set 8k */
             )
         );
 
-        final Duration delay = Duration.ofMillis(1);
         final StreamingRequest<ByteBuffer> streamingRequest = new StreamingRequest<>(
             "POST",
             "/_bulk/stream",

plugins/transport-reactor-netty4/src/main/java/org/opensearch/http/reactor/netty4/ReactorNetty4HttpServerTransport.java

@@ -230,6 +230,7 @@ protected HttpServerChannel bind(InetSocketAddress socketAddress) throws Excepti
                     spec -> spec.maxChunkSize(maxChunkSize.bytesAsInt())
                         .maxHeaderSize(maxHeaderSize.bytesAsInt())
                         .maxInitialLineLength(maxInitialLineLength.bytesAsInt())
+                        .allowPartialChunks(false)
                 )
                 .handle((req, res) -> incomingRequest(req, res))
         );

rest-api-spec/src/main/resources/rest-api-spec/test/indices.create/10_basic.yml

@@ -112,3 +112,33 @@
             properties:
               "":
                 type:     keyword
+
+---
+"Create index with setting index.number_of_routing_shards to null":
+  - skip:
+      version: " - 2.17.99"
+      reason: "fixed in 2.18.0"
+  - do:
+      indices.create:
+        index: test_index
+        body:
+          settings:
+            number_of_routing_shards: null
+  - do:
+      cluster.state:
+        metric: [ metadata ]
+        index: test_index
+  - match : { metadata.indices.test_index.routing_num_shards: 1024 }
+
+  - do:
+      indices.create:
+        index: test_index1
+        body:
+          settings:
+            number_of_routing_shards: null
+            number_of_shards: 3
+  - do:
+      cluster.state:
+        metric: [ metadata ]
+        index: test_index1
+  - match : { metadata.indices.test_index1.routing_num_shards: 768 }