Merge branch 'main' into main

Author: navneet1v

.ci/bwcVersions

@@ -57,3 +57,4 @@ BWC_VERSION:
   - "3.3.2"
   - "3.4.0"
   - "3.5.0"
+  - "3.6.0"

.gitignore

@@ -1,6 +1,7 @@
 .claude
 CLAUDE.md
 .cursor*
+.kiro*
 
 # intellij files
 .idea/

CHANGELOG.md

@@ -25,6 +25,8 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Add new setting property 'Sensitive' for tiering dynamic settings ([#20901](https://github.com/opensearch-project/OpenSearch/pull/20901))
 - Add ref_path support for package-based hunspell dictionary loading ([#20840](https://github.com/opensearch-project/OpenSearch/pull/20840))
 - Add support for enabling pluggable data formats, starting with phase-1 of decoupling shard from engine, and introducing basic abstractions ([#20675](https://github.com/opensearch-project/OpenSearch/pull/20675))
+- Add concurrent queue in libs and composite engine sandbox plugin ([#20909](https://github.com/opensearch-project/OpenSearch/pull/20909))
+- Add interface for the Multi format merge flow ([#20908](https://github.com/opensearch-project/OpenSearch/pull/20908))
 
 - Add warmup phase to wait for lag to catch up in pull-based ingestion before serving ([#20526](https://github.com/opensearch-project/OpenSearch/pull/20526))
 - Add a new static method to IndicesOptions API to expose `STRICT_EXPAND_OPEN_HIDDEN_FORBID_CLOSED` index option ([#20980](https://github.com/opensearch-project/OpenSearch/pull/20980))
@@ -97,4 +99,4 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 
 ### Removed
 
-[Unreleased 3.x]: https://github.com/opensearch-project/OpenSearch/compare/3.6...main
+[Unreleased 3.x]: https://github.com/opensearch-project/OpenSearch/compare/3.7...main

buildSrc/version.properties

@@ -1,2 +1,2 @@
 # Please use ../gradle/libs.versions.toml for dependency management
-opensearch        = 3.6.0
+opensearch        = 3.7.0

gradle/libs.versions.toml

@@ -1,5 +1,5 @@
 [versions]
-opensearch        = "3.6.0"
+opensearch        = "3.7.0"
 lucene            = "10.4.0"
 
 bundled_jdk_vendor = "adoptium"

libs/concurrent-queue/build.gradle

@@ -0,0 +1,71 @@
+/*
+ * 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.
+ */
+
+/*
+ * Shared concurrent queue utilities for the composite indexing engine.
+ * No external dependencies — pure Java concurrency primitives.
+ */
+
+// ---- JMH benchmark source set ----
+sourceSets {
+  jmh {
+    java.srcDirs = ['src/jmh/java']
+    compileClasspath += sourceSets.main.output
+    runtimeClasspath += sourceSets.main.output
+  }
+}
+
+dependencies {
+  /*******
+   *  !!!! NO RUNTIME DEPENDENCIES !!!!
+   *******/
+
+  testImplementation "com.carrotsearch.randomizedtesting:randomizedtesting-runner:${versions.randomizedrunner}"
+  testImplementation "junit:junit:${versions.junit}"
+  testImplementation "org.hamcrest:hamcrest:${versions.hamcrest}"
+
+  testImplementation(project(":test:framework")) {
+    exclude group: 'org.opensearch', module: 'opensearch-concurrent-queue'
+  }
+
+  // JMH benchmark dependencies
+  jmhImplementation "org.openjdk.jmh:jmh-core:${versions.jmh}"
+  jmhAnnotationProcessor "org.openjdk.jmh:jmh-generator-annprocess:${versions.jmh}"
+}
+
+testingConventions.enabled = true
+
+tasks.named('forbiddenApisMain').configure {
+  // :libs:opensearch-concurrent-queue does not depend on server
+  replaceSignatureFiles 'jdk-signatures'
+}
+
+// JMH generated code uses forbidden APIs — skip checking
+tasks.matching { it.name == 'forbiddenApisJmh' }.configureEach { enabled = false }
+
+// JMH annotation processor
+tasks.named('compileJmhJava').configure {
+  options.compilerArgs.addAll(["-processor", "org.openjdk.jmh.generators.BenchmarkProcessor"])
+}
+
+// Convenience task to run JMH benchmarks
+tasks.register('jmh', JavaExec) {
+  dependsOn jmhClasses
+  mainClass = 'org.openjdk.jmh.Main'
+  classpath = sourceSets.jmh.runtimeClasspath
+  // Pass -Pjmh.includes=<regex> to filter benchmarks
+  if (project.hasProperty('jmh.includes')) {
+    args project.property('jmh.includes')
+  }
+}
+
+spotless {
+  java {
+    targetExclude 'src/jmh/generated/**/*.java'
+  }
+}

libs/concurrent-queue/src/jmh/java/org/opensearch/common/queue/LockableConcurrentQueueBenchmark.java

@@ -0,0 +1,141 @@
+/*
+ * 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.common.queue;
+
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Threads;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.infra.Blackhole;
+
+import java.util.LinkedList;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * JMH benchmark for {@link LockableConcurrentQueue} measuring throughput of
+ * lock-and-poll / add-and-unlock cycles under varying concurrency levels.
+ */
+@Fork(3)
+@Warmup(iterations = 5, time = 1)
+@Measurement(iterations = 10, time = 1)
+@BenchmarkMode(Mode.Throughput)
+@OutputTimeUnit(TimeUnit.MICROSECONDS)
+@State(Scope.Benchmark)
+@SuppressWarnings("unused")
+public class LockableConcurrentQueueBenchmark {
+
+    @Param({ "1", "4", "8" })
+    int concurrency;
+
+    @Param({ "16", "64" })
+    int poolSize;
+
+    private LockableConcurrentQueue<LockableEntry> queue;
+
+    @Setup(Level.Iteration)
+    public void setup() {
+        queue = new LockableConcurrentQueue<>(LinkedList::new, concurrency);
+        for (int i = 0; i < poolSize; i++) {
+            LockableEntry entry = new LockableEntry();
+            entry.lock();
+            queue.addAndUnlock(entry);
+        }
+    }
+
+    @Benchmark
+    @Threads(1)
+    public LockableEntry pollAndReturn_1thread() {
+        return pollAndReturn();
+    }
+
+    @Benchmark
+    @Threads(4)
+    public LockableEntry pollAndReturn_4threads() {
+        return pollAndReturn();
+    }
+
+    @Benchmark
+    @Threads(8)
+    public LockableEntry pollAndReturn_8threads() {
+        return pollAndReturn();
+    }
+
+    private LockableEntry pollAndReturn() {
+        LockableEntry entry = queue.lockAndPoll();
+        if (entry != null) {
+            queue.addAndUnlock(entry);
+        }
+        return entry;
+    }
+
+    @Benchmark
+    @Threads(4)
+    public void writerWorkload_4threads(Blackhole bh) {
+        writerWorkload(bh);
+    }
+
+    @Benchmark
+    @Threads(8)
+    public void writerWorkload_8threads(Blackhole bh) {
+        writerWorkload(bh);
+    }
+
+    @Benchmark
+    @Threads(16)
+    public void writerWorkload_16threads(Blackhole bh) {
+        writerWorkload(bh);
+    }
+
+    private void writerWorkload(Blackhole bh) {
+        LockableEntry entry = queue.lockAndPoll();
+        if (entry != null) {
+            bh.consume(simulateWork(entry));
+            queue.addAndUnlock(entry);
+        }
+    }
+
+    private static long simulateWork(LockableEntry entry) {
+        long result = entry.hashCode();
+        for (int i = 0; i < 20; i++) {
+            result ^= (result << 13);
+            result ^= (result >> 7);
+            result ^= (result << 17);
+        }
+        return result;
+    }
+
+    static final class LockableEntry implements Lockable {
+        private final ReentrantLock lock = new ReentrantLock();
+
+        @Override
+        public void lock() {
+            lock.lock();
+        }
+
+        @Override
+        public boolean tryLock() {
+            return lock.tryLock();
+        }
+
+        @Override
+        public void unlock() {
+            lock.unlock();
+        }
+    }
+}