Merge remote-tracking branch 'origin/main' into dev/fix-PemTrustConfigTests-testTrustConfigReloadsFileContents

Author: guojialiang92

CHANGELOG.md

@@ -5,6 +5,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 
 ## [Unreleased 3.x]
 ### Added
+- Add multi-threaded writer support in pull-based ingestion ([#17912](https://github.com/opensearch-project/OpenSearch/pull/17912))
 
 ### Changed
 

build.gradle

@@ -55,6 +55,7 @@ plugins {
   id 'opensearch.docker-support'
   id 'opensearch.global-build-info'
   id "com.diffplug.spotless" version "6.25.0" apply false
+  id "org.gradle.test-retry" version "1.6.2" apply false
   id "test-report-aggregation"
   id 'jacoco-report-aggregation'
 }
@@ -70,13 +71,6 @@ apply from: 'gradle/run.gradle'
 apply from: 'gradle/missing-javadoc.gradle'
 apply from: 'gradle/code-coverage.gradle'
 
-// Disable unconditional publishing of build scans
-develocity {
-  buildScan {
-    publishing.onlyIf { false }
-  }
-}
-
 // common maven publishing configuration
 allprojects {
   group = 'org.opensearch'
@@ -475,8 +469,9 @@ gradle.projectsEvaluated {
 
 // test retry configuration
 subprojects {
+  apply plugin: "org.gradle.test-retry"
   tasks.withType(Test).configureEach {
-    develocity.testRetry {
+    retry {
       if (BuildParams.isCi()) {
         maxRetries = 3
         maxFailures = 10

gradle.properties

@@ -21,9 +21,6 @@ org.gradle.jvmargs=-Xmx3g -XX:+HeapDumpOnOutOfMemoryError -Xss2m \
     --add-exports jdk.compiler/com.sun.tools.javac.util=ALL-UNNAMED
 options.forkOptions.memoryMaximumSize=3g
 
-# Disable Gradle Enterprise Gradle plugin's test retry
-systemProp.develocity.testretry.enabled.enabled=false
-
 # Disable duplicate project id detection
 # See https://docs.gradle.org/current/userguide/upgrading_version_6.html#duplicate_project_names_may_cause_publication_to_fail
 systemProp.org.gradle.dependency.duplicate.project.detection=false

libs/agent-sm/agent/src/main/java/org/opensearch/javaagent/Agent.java

@@ -11,6 +11,7 @@
 import org.opensearch.javaagent.bootstrap.AgentPolicy;
 
 import java.lang.instrument.Instrumentation;
+import java.net.Socket;
 import java.nio.channels.FileChannel;
 import java.nio.channels.SocketChannel;
 import java.nio.file.Files;
@@ -71,8 +72,9 @@ public static void agentmain(String agentArguments, Instrumentation instrumentat
         initAgent(instrumentation);
     }
 
-    private static AgentBuilder createAgentBuilder(Instrumentation inst) throws Exception {
-        final Junction<TypeDescription> systemType = ElementMatchers.isSubTypeOf(SocketChannel.class);
+    private static AgentBuilder createAgentBuilder() throws Exception {
+        final Junction<TypeDescription> socketType = ElementMatchers.isSubTypeOf(SocketChannel.class)
+            .or(ElementMatchers.isSubTypeOf(Socket.class));
         final Junction<TypeDescription> pathType = ElementMatchers.isSubTypeOf(Files.class);
         final Junction<TypeDescription> fileChannelType = ElementMatchers.isSubTypeOf(FileChannel.class);
 
@@ -98,11 +100,11 @@ private static AgentBuilder createAgentBuilder(Instrumentation inst) throws Exce
             );
 
         final ByteBuddy byteBuddy = new ByteBuddy().with(Implementation.Context.Disabled.Factory.INSTANCE);
-        final AgentBuilder agentBuilder = new AgentBuilder.Default(byteBuddy).with(AgentBuilder.InitializationStrategy.NoOp.INSTANCE)
+        return new AgentBuilder.Default(byteBuddy).with(AgentBuilder.InitializationStrategy.NoOp.INSTANCE)
             .with(AgentBuilder.RedefinitionStrategy.REDEFINITION)
             .with(AgentBuilder.TypeStrategy.Default.REDEFINE)
             .ignore(ElementMatchers.nameContains("$MockitoMock$")) /* ingore all Mockito mocks */
-            .type(systemType)
+            .type(socketType)
             .transform(socketTransformer)
             .type(pathType.or(fileChannelType))
             .transform(fileTransformer)
@@ -118,12 +120,10 @@ private static AgentBuilder createAgentBuilder(Instrumentation inst) throws Exce
                     Advice.to(RuntimeHaltInterceptor.class).on(ElementMatchers.named("halt"))
                 )
             );
-
-        return agentBuilder;
     }
 
     private static void initAgent(Instrumentation instrumentation) throws Exception {
-        AgentBuilder agentBuilder = createAgentBuilder(instrumentation);
+        AgentBuilder agentBuilder = createAgentBuilder();
         agentBuilder.installOn(instrumentation);
     }
 }

libs/agent-sm/agent/src/test/java/org/opensearch/javaagent/SocketChannelInterceptorTests.java

@@ -13,6 +13,7 @@
 import java.io.IOException;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
+import java.net.Socket;
 import java.net.UnixDomainSocketAddress;
 import java.nio.channels.SocketChannel;
 
@@ -28,6 +29,8 @@ public void testConnections() throws IOException {
 
             assertThrows(SecurityException.class, () -> channel.connect(new InetSocketAddress("opensearch.org", 80)));
         }
+
+        assertThrows(SecurityException.class, () -> new Socket("localhost", 9200));
     }
 
     @Test

modules/repository-url/src/test/resources/org/opensearch/bootstrap/test.policy

@@ -0,0 +1,11 @@
+/*
+ * 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.
+ */
+
+grant {
+  permission java.net.SocketPermission "*", "connect";
+};

plugins/discovery-ec2/src/test/resources/org/opensearch/bootstrap/test.policy

@@ -0,0 +1,11 @@
+/*
+ * 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.
+ */
+
+grant {
+  permission java.net.SocketPermission "*", "connect";
+};

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

@@ -200,4 +200,31 @@ public void testUpdateWithoutIDField() throws Exception {
             return 30 == (Integer) response.getHits().getHits()[0].getSourceAsMap().get("age");
         });
     }
+
+    public void testMultiThreadedWrites() throws Exception {
+        // create index with 5 writer threads
+        createIndexWithDefaultSettings(indexName, 1, 0, 5);
+        ensureGreen(indexName);
+
+        // Step 1: Produce messages
+        for (int i = 0; i < 1000; i++) {
+            produceData(Integer.toString(i), "name" + i, "25");
+        }
+
+        waitForState(() -> {
+            SearchResponse searchableDocsResponse = client().prepareSearch(indexName).setSize(2000).setPreference("_only_local").get();
+            return searchableDocsResponse.getHits().getTotalHits().value() == 1000;
+        });
+
+        // Step 2: Produce an update message and validate
+        for (int i = 0; i < 1000; i++) {
+            produceData(Integer.toString(i), "name" + i, "30");
+        }
+
+        waitForState(() -> {
+            RangeQueryBuilder query = new RangeQueryBuilder("age").gte(28);
+            SearchResponse response = client().prepareSearch(indexName).setQuery(query).get();
+            return response.getHits().getTotalHits().value() == 1000;
+        });
+    }
 }

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

@@ -177,10 +177,10 @@ protected ResumeIngestionResponse resumeIngestion(String indexName) throws Execu
     }
 
     protected void createIndexWithDefaultSettings(int numShards, int numReplicas) {
-        createIndexWithDefaultSettings(indexName, numShards, numReplicas);
+        createIndexWithDefaultSettings(indexName, numShards, numReplicas, 1);
     }
 
-    protected void createIndexWithDefaultSettings(String indexName, int numShards, int numReplicas) {
+    protected void createIndexWithDefaultSettings(String indexName, int numShards, int numReplicas, int numProcessorThreads) {
         createIndex(
             indexName,
             Settings.builder()
@@ -191,6 +191,7 @@ protected void createIndexWithDefaultSettings(String indexName, int numShards, i
                 .put("ingestion_source.param.topic", topicName)
                 .put("ingestion_source.param.bootstrap_servers", kafka.getBootstrapServers())
                 .put("index.replication.type", "SEGMENT")
+                .put("ingestion_source.num_processor_threads", numProcessorThreads)
                 // set custom kafka consumer properties
                 .put("ingestion_source.param.fetch.min.bytes", 30000)
                 .put("ingestion_source.param.enable.auto.commit", false)

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

@@ -164,6 +164,7 @@ public void testErrorStrategy() throws Exception {
             .setSettings(Settings.builder().put("ingestion_source.error_strategy", "drop"))
             .get();
         waitForState(() -> "drop".equalsIgnoreCase(getSettings(indexName, "index.ingestion_source.error_strategy")));
+        resumeIngestion(indexName);
         waitForSearchableDocs(2, Arrays.asList(node));
     }
 
@@ -248,8 +249,8 @@ public void testPaginatedGetIngestionState() throws ExecutionException, Interrup
         internalCluster().startClusterManagerOnlyNode();
         internalCluster().startDataOnlyNode();
         internalCluster().startDataOnlyNode();
-        createIndexWithDefaultSettings("index1", 5, 0);
-        createIndexWithDefaultSettings("index2", 5, 0);
+        createIndexWithDefaultSettings("index1", 5, 0, 1);
+        createIndexWithDefaultSettings("index2", 5, 0, 1);
         ensureGreen("index1");
         ensureGreen("index2");