Merge branch 'main' into dependabot/gradle/reactor_netty-1.2.3

Signed-off-by: gaobinlong <[email protected]>

Author: gaobinlong

.github/workflows/changelog_verifier.yml

@@ -14,42 +14,5 @@ jobs:
           token: ${{ secrets.GITHUB_TOKEN }}
           ref: ${{ github.event.pull_request.head.sha }}
       - uses: dangoslen/changelog-enforcer@v3
-        id: verify-changelog-3x
         with:
           skipLabels: "autocut, skip-changelog"
-          changeLogPath: 'CHANGELOG-3.0.md'
-        continue-on-error: true
-      - uses: dangoslen/changelog-enforcer@v3
-        id: verify-changelog
-        with:
-          skipLabels: "autocut, skip-changelog"
-          changeLogPath: 'CHANGELOG.md'
-        continue-on-error: true
-      - run: |
-          # The check was possibly skipped leading to success for both the jobs
-          has_backport_label=${{ contains(join(github.event.pull_request.labels.*.name, ', '), 'backport')}}
-          has_breaking_label=${{ contains(join(github.event.pull_request.labels.*.name, ', '), '>breaking')}}
-          if [[ $has_breaking_label == true && $has_backport_label == true ]]; then
-            echo "error: Please make sure that the PR does not have a backport label associated with it when making breaking changes"
-            exit 1
-          fi
-
-          if [[ ${{ steps.verify-changelog-3x.outcome }} == 'success' && ${{ steps.verify-changelog.outcome }} == 'success' ]]; then
-            exit 0
-          fi
-
-          if [[ ${{ steps.verify-changelog-3x.outcome }} == 'failure' && ${{ steps.verify-changelog.outcome }} == 'failure' ]]; then
-            echo "error: Please ensure a changelog entry exists in CHANGELOG.md or CHANGELOG-3.0.md"
-            exit 1
-          fi
-
-          # Concatenates the labels and checks if the string contains "backport"
-          if [[ ${{ steps.verify-changelog.outcome }} == 'success' && $has_backport_label == false ]]; then
-            echo "error: Please make sure that the PR has a backport label associated with it when making an entry to the CHANGELOG.md file"
-            exit 1
-          fi
-
-          if [[ ${{ steps.verify-changelog-3x.outcome }} == 'success' && $has_backport_label == true ]]; then
-            echo "error: Please make sure that the PR does not have a backport label associated with it when making an entry to the CHANGELOG-3.0.md file"
-            exit 1
-          fi

CHANGELOG-3.0.md

@@ -3,11 +3,16 @@ All notable changes to this project are documented in this file.
 
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). See the [CONTRIBUTING guide](./CONTRIBUTING.md#Changelog) for instructions on how to add changelog entries.
 
-## [Unreleased 2.x]
+## [Unreleased 3.x]
 ### Added
+- Change priority for scheduling reroute during timeout([#16445](https://github.com/opensearch-project/OpenSearch/pull/16445))
 - Renaming the node role search to warm ([#17573](https://github.com/opensearch-project/OpenSearch/pull/17573))
+- Introduce a new search node role to hold search only shards ([#17620](https://github.com/opensearch-project/OpenSearch/pull/17620))
+
 ### Dependencies
 - Bump `reactor_netty` from 1.1.26 to 1.2.3 ([#17322](https://github.com/opensearch-project/OpenSearch/pull/17322), [#17377](https://github.com/opensearch-project/OpenSearch/pull/17377))
+- Bump `ch.qos.logback:logback-core` from 1.5.16 to 1.5.17 ([#17609](https://github.com/opensearch-project/OpenSearch/pull/17609))
+- Bump `org.jruby.joni:joni` from 2.2.3 to 2.2.5 ([#17608](https://github.com/opensearch-project/OpenSearch/pull/17608))
 
 ### Changed
 
@@ -19,4 +24,4 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 
 ### Security
 
-[Unreleased 2.x]: https://github.com/opensearch-project/OpenSearch/compare/2.19...2.x
+[Unreleased 2.x]: https://github.com/opensearch-project/OpenSearch/compare/f58d846f...main

CHANGELOG.md

@@ -146,15 +146,6 @@ Adding in the change is two step process:
 1. Add your changes to the corresponding section within the CHANGELOG file with dummy pull request information, publish the PR
 2. Update the entry for your change in [`CHANGELOG.md`](CHANGELOG.md) and make sure that you reference the pull request there.
 
-### Where should I put my CHANGELOG entry?
-Please review the [branching strategy](https://github.com/opensearch-project/.github/blob/main/RELEASING.md#opensearch-branching) document. The changelog on the `main` branch will contain **two files**: `CHANGELOG.md` which corresponds to unreleased changes intended for the _next minor_ release and `CHANGELOG-3.0.md` which correspond to unreleased changes intended for the _next major_ release. Your entry should go into file corresponding to the version it is intended to be released in. In practice, most changes to `main` will be backported to the next minor release so most entries will be in the `CHANGELOG.md` file.
-
-The following examples assume the _next major_ release on main is 3.0, then _next minor_ release is 2.5, and the _current_ release is 2.4.
-
-- **Add a new feature to release in next minor:** Add a changelog entry to `[Unreleased 2.x]` in CHANGELOG.md on main, then backport to 2.x (including the changelog entry).
-- **Introduce a breaking API change to release in next major:** Add a changelog entry to `[Unreleased 3.0]` to CHANGELOG-3.0.md on main, do not backport.
-- **Upgrade a dependency to fix a CVE:** Add a changelog entry to `[Unreleased 2.x]` on main, then backport to 2.x (including the changelog entry), then backport to 2.4 and ensure the changelog entry is added to `[Unreleased 2.4.1]`.
-
 ## Review Process
 
 We deeply appreciate everyone who takes the time to make a contribution. We will review all contributions as quickly as possible. As a reminder, [opening an issue](https://github.com/opensearch-project/OpenSearch/issues/new/choose) discussing your change before you make it is the best way to smooth the PR process. This will prevent a rejection because someone else is already working on the problem, or because the solution is incompatible with the architectural direction.

CONTRIBUTING.md

@@ -29,7 +29,7 @@
  */
 
 dependencies {
-  api 'org.jruby.joni:joni:2.2.3'
+  api 'org.jruby.joni:joni:2.2.5'
   // joni dependencies:
   api 'org.jruby.jcodings:jcodings:1.0.63'
 

libs/grok/build.gradle

@@ -0,0 +1 @@
+4ebafe67efa7395678a34d07e7585bed5ef0cc72

libs/grok/licenses/joni-2.2.3.jar.sha1

@@ -34,7 +34,7 @@
 import org.testcontainers.utility.DockerImageName;
 
 /**
- * Base test class for Kafka ingestion tests
+ * Base test class for Kafka ingestion tests.
  */
 @ThreadLeakFilters(filters = TestContainerThreadLeakFilter.class)
 public class KafkaIngestionBaseIT extends OpenSearchIntegTestCase {
@@ -135,6 +135,9 @@ protected void createIndexWithDefaultSettings(int numShards, int numReplicas) {
                 .put("ingestion_source.param.topic", topicName)
                 .put("ingestion_source.param.bootstrap_servers", kafka.getBootstrapServers())
                 .put("index.replication.type", "SEGMENT")
+                // set custom kafka consumer properties
+                .put("ingestion_source.param.fetch.min.bytes", 30000)
+                .put("ingestion_source.param.enable.auto.commit", false)
                 .build(),
             "{\"properties\":{\"name\":{\"type\": \"text\"},\"age\":{\"type\": \"integer\"}}}}"
         );

libs/grok/licenses/joni-2.2.5.jar.sha1

@@ -48,7 +48,6 @@ public void testSegmentReplicationWithRemoteStore() throws Exception {
         internalCluster().startClusterManagerOnlyNode();
         final String nodeA = internalCluster().startDataOnlyNode();
         createIndexWithDefaultSettings(1, 1);
-
         ensureYellowAndNoInitializingShards(indexName);
         final String nodeB = internalCluster().startDataOnlyNode();
         ensureGreen(indexName);

plugins/ingestion-kafka/src/internalClusterTest/java/org/opensearch/plugin/kafka/KafkaIngestionBaseIT.java

@@ -9,7 +9,6 @@
 package org.opensearch.plugin.kafka;
 
 import org.apache.kafka.clients.consumer.Consumer;
-import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
@@ -99,9 +98,10 @@ protected static Consumer<byte[], byte[]> createConsumer(String clientId, KafkaS
         Properties consumerProp = new Properties();
         consumerProp.put("bootstrap.servers", config.getBootstrapServers());
         consumerProp.put("client.id", clientId);
-        if (config.getAutoOffsetResetConfig() != null && !config.getAutoOffsetResetConfig().isEmpty()) {
-            consumerProp.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, config.getAutoOffsetResetConfig());
-        }
+
+        logger.info("Kafka consumer properties for topic {}: {}", config.getTopic(), config.getConsumerConfigurations());
+        consumerProp.putAll(config.getConsumerConfigurations());
+
         // TODO: why Class org.apache.kafka.common.serialization.StringDeserializer could not be found if set the deserializer as prop?
         // consumerProp.put("key.deserializer",
         // "org.apache.kafka.common.serialization.StringDeserializer");

plugins/ingestion-kafka/src/internalClusterTest/java/org/opensearch/plugin/kafka/RemoteStoreKafkaIT.java

@@ -10,6 +10,7 @@
 
 import org.opensearch.core.util.ConfigurationUtils;
 
+import java.util.HashMap;
 import java.util.Map;
 
 /**
@@ -18,21 +19,27 @@
 public class KafkaSourceConfig {
     private final String PROP_TOPIC = "topic";
     private final String PROP_BOOTSTRAP_SERVERS = "bootstrap_servers";
-    // TODO: support pass any generic kafka configs
     private final String PROP_AUTO_OFFSET_RESET = "auto.offset.reset";
 
     private final String topic;
     private final String bootstrapServers;
     private final String autoOffsetResetConfig;
 
+    private final Map<String, Object> consumerConfigsMap;
+
     /**
-     * Constructor
+     * Extracts and look for required and optional kafka consumer configurations.
      * @param params the configuration parameters
      */
     public KafkaSourceConfig(Map<String, Object> params) {
         this.topic = ConfigurationUtils.readStringProperty(params, PROP_TOPIC);
         this.bootstrapServers = ConfigurationUtils.readStringProperty(params, PROP_BOOTSTRAP_SERVERS);
         this.autoOffsetResetConfig = ConfigurationUtils.readOptionalStringProperty(params, PROP_AUTO_OFFSET_RESET);
+        this.consumerConfigsMap = new HashMap<>(params);
+
+        // remove above configurations
+        consumerConfigsMap.remove(PROP_TOPIC);
+        consumerConfigsMap.remove(PROP_BOOTSTRAP_SERVERS);
     }
 
     /**
@@ -60,4 +67,8 @@ public String getBootstrapServers() {
     public String getAutoOffsetResetConfig() {
         return autoOffsetResetConfig;
     }
+
+    public Map<String, Object> getConsumerConfigurations() {
+        return consumerConfigsMap;
+    }
 }

plugins/ingestion-kafka/src/main/java/org/opensearch/plugin/kafka/KafkaPartitionConsumer.java

@@ -16,10 +16,12 @@
 
 public class KafkaSourceConfigTests extends OpenSearchTestCase {
 
-    public void testConstructorAndGetters() {
+    public void testKafkaSourceConfig() {
         Map<String, Object> params = new HashMap<>();
         params.put("topic", "topic");
         params.put("bootstrap_servers", "bootstrap");
+        params.put("fetch.min.bytes", 30000);
+        params.put("enable.auto.commit", false);
 
         KafkaSourceConfig config = new KafkaSourceConfig(params);
 
@@ -29,5 +31,7 @@ public void testConstructorAndGetters() {
             "bootstrap",
             config.getBootstrapServers()
         );
+        Assert.assertEquals("Incorrect fetch.min.bytes", 30000, config.getConsumerConfigurations().get("fetch.min.bytes"));
+        Assert.assertEquals("Incorrect enable.auto.commit", false, config.getConsumerConfigurations().get("enable.auto.commit"));
     }
 }

plugins/ingestion-kafka/src/main/java/org/opensearch/plugin/kafka/KafkaSourceConfig.java

@@ -90,7 +90,7 @@ private void waitForNodes(int numNodes) {
     public void testNodeCounts() {
         int total = 1;
         internalCluster().startNode();
-        Map<String, Integer> expectedCounts = getExpectedCounts(1, 1, 1, 1, 1, 0, 0);
+        Map<String, Integer> expectedCounts = getExpectedCounts(1, 1, 1, 1, 1, 0, 0, 0);
         int numNodes = randomIntBetween(1, 5);
 
         ClusterStatsResponse response = client().admin()
@@ -159,7 +159,7 @@ public void testNodeCountsWithDeprecatedMasterRole() throws ExecutionException,
         internalCluster().startNode(settings);
         waitForNodes(total);
 
-        Map<String, Integer> expectedCounts = getExpectedCounts(0, 1, 1, 0, 0, 0, 0);
+        Map<String, Integer> expectedCounts = getExpectedCounts(0, 1, 1, 0, 0, 0, 0, 0);
 
         Client client = client();
         ClusterStatsResponse response = client.admin()
@@ -484,7 +484,7 @@ public void testNodeRolesWithMasterLegacySettings() throws ExecutionException, I
         internalCluster().startNodes(legacyMasterSettings);
         waitForNodes(total);
 
-        Map<String, Integer> expectedCounts = getExpectedCounts(0, 1, 1, 0, 1, 0, 0);
+        Map<String, Integer> expectedCounts = getExpectedCounts(0, 1, 1, 0, 1, 0, 0, 0);
 
         Client client = client();
         ClusterStatsResponse clusterStatsResponse = client.admin()
@@ -518,7 +518,7 @@ public void testNodeRolesWithClusterManagerRole() throws ExecutionException, Int
         internalCluster().startNodes(clusterManagerNodeRoleSettings);
         waitForNodes(total);
 
-        Map<String, Integer> expectedCounts = getExpectedCounts(0, 1, 1, 0, 1, 0, 0);
+        Map<String, Integer> expectedCounts = getExpectedCounts(0, 1, 1, 0, 1, 0, 0, 0);
 
         Client client = client();
         ClusterStatsResponse clusterStatsResponse = client.admin()
@@ -546,7 +546,7 @@ public void testNodeRolesWithSeedDataNodeLegacySettings() throws ExecutionExcept
         internalCluster().startNodes(legacySeedDataNodeSettings);
         waitForNodes(total);
 
-        Map<String, Integer> expectedRoleCounts = getExpectedCounts(1, 1, 1, 0, 1, 0, 0);
+        Map<String, Integer> expectedRoleCounts = getExpectedCounts(1, 1, 1, 0, 1, 0, 0, 0);
 
         Client client = client();
         ClusterStatsResponse clusterStatsResponse = client.admin()
@@ -577,7 +577,7 @@ public void testNodeRolesWithDataNodeLegacySettings() throws ExecutionException,
         internalCluster().startNodes(legacyDataNodeSettings);
         waitForNodes(total);
 
-        Map<String, Integer> expectedRoleCounts = getExpectedCounts(1, 1, 1, 0, 1, 0, 0);
+        Map<String, Integer> expectedRoleCounts = getExpectedCounts(1, 1, 1, 0, 1, 0, 0, 0);
 
         Client client = client();
         ClusterStatsResponse clusterStatsResponse = client.admin()
@@ -594,6 +594,29 @@ public void testNodeRolesWithDataNodeLegacySettings() throws ExecutionException,
         assertEquals(expectedNodesRoles, Set.of(getNodeRoles(client, 0), getNodeRoles(client, 1)));
     }
 
+    public void testNodeRolesWithSearchNode() throws ExecutionException, InterruptedException {
+        int total = 2;
+        internalCluster().startClusterManagerOnlyNodes(1);
+        internalCluster().startSearchOnlyNode();
+        waitForNodes(total);
+
+        Map<String, Integer> expectedRoleCounts = getExpectedCounts(0, 1, 1, 0, 0, 0, 1, 0);
+
+        Client client = client();
+        ClusterStatsResponse clusterStatsResponse = client.admin()
+            .cluster()
+            .prepareClusterStats()
+            .useAggregatedNodeLevelResponses(randomBoolean())
+            .get();
+        assertCounts(clusterStatsResponse.getNodesStats().getCounts(), total, expectedRoleCounts);
+
+        Set<Set<String>> expectedNodesRoles = Set.of(
+            Set.of(DiscoveryNodeRole.CLUSTER_MANAGER_ROLE.roleName()),
+            Set.of(DiscoveryNodeRole.SEARCH_ROLE.roleName())
+        );
+        assertEquals(expectedNodesRoles, Set.of(getNodeRoles(client, 0), getNodeRoles(client, 1)));
+    }
+
     public void testClusterStatsWithNodeMetricsFilter() {
         internalCluster().startNode();
         ensureGreen();
@@ -887,6 +910,7 @@ private Map<String, Integer> getExpectedCounts(
         int clusterManagerRoleCount,
         int ingestRoleCount,
         int remoteClusterClientRoleCount,
+        int warmRoleCount,
         int searchRoleCount,
         int coordinatingOnlyCount
     ) {
@@ -896,7 +920,8 @@ private Map<String, Integer> getExpectedCounts(
         expectedCounts.put(DiscoveryNodeRole.CLUSTER_MANAGER_ROLE.roleName(), clusterManagerRoleCount);
         expectedCounts.put(DiscoveryNodeRole.INGEST_ROLE.roleName(), ingestRoleCount);
         expectedCounts.put(DiscoveryNodeRole.REMOTE_CLUSTER_CLIENT_ROLE.roleName(), remoteClusterClientRoleCount);
-        expectedCounts.put(DiscoveryNodeRole.WARM_ROLE.roleName(), searchRoleCount);
+        expectedCounts.put(DiscoveryNodeRole.WARM_ROLE.roleName(), warmRoleCount);
+        expectedCounts.put(DiscoveryNodeRole.SEARCH_ROLE.roleName(), searchRoleCount);
         expectedCounts.put(ClusterStatsNodes.Counts.COORDINATING_ONLY, coordinatingOnlyCount);
         return expectedCounts;
     }

plugins/ingestion-kafka/src/test/java/org/opensearch/plugin/kafka/KafkaSourceConfigTests.java

@@ -488,6 +488,15 @@ public boolean isWarmNode() {
         return roles.contains(DiscoveryNodeRole.WARM_ROLE);
     }
 
+    /**
+     * Returns whether the node is dedicated to host search replicas.
+     *
+     * @return true if the node contains a search role, false otherwise
+     */
+    public boolean isSearchNode() {
+        return roles.contains(DiscoveryNodeRole.SEARCH_ROLE);
+    }
+
     /**
      * Returns whether the node is a remote store node.
      *

server/src/internalClusterTest/java/org/opensearch/action/admin/cluster/stats/ClusterStatsIT.java

@@ -310,11 +310,39 @@ public Setting<Boolean> legacySetting() {
 
     };
 
+    /**
+     * Represents the role for a search node, which is dedicated to host search replicas.
+     */
+    public static final DiscoveryNodeRole SEARCH_ROLE = new DiscoveryNodeRole("search", "s", true) {
+
+        @Override
+        public Setting<Boolean> legacySetting() {
+            // search role is added in 2.4 so doesn't need to configure legacy setting
+            return null;
+        }
+
+        @Override
+        public void validateRole(List<DiscoveryNodeRole> roles) {
+            for (DiscoveryNodeRole role : roles) {
+                if (role.equals(DiscoveryNodeRole.SEARCH_ROLE) == false) {
+                    throw new IllegalArgumentException(
+                        String.format(
+                            Locale.ROOT,
+                            "%s role cannot be combined with any other role on a node.",
+                            DiscoveryNodeRole.SEARCH_ROLE.roleName()
+                        )
+                    );
+                }
+            }
+        }
+
+    };
+
     /**
      * The built-in node roles.
      */
     public static SortedSet<DiscoveryNodeRole> BUILT_IN_ROLES = Collections.unmodifiableSortedSet(
-        new TreeSet<>(Arrays.asList(DATA_ROLE, INGEST_ROLE, CLUSTER_MANAGER_ROLE, REMOTE_CLUSTER_CLIENT_ROLE, WARM_ROLE))
+        new TreeSet<>(Arrays.asList(DATA_ROLE, INGEST_ROLE, CLUSTER_MANAGER_ROLE, REMOTE_CLUSTER_CLIENT_ROLE, WARM_ROLE, SEARCH_ROLE))
     );
 
     /**