Add sandbox plugin for composite engine

Signed-off-by: Arpit Bandejiya <abandeji@amazon.com>

Author: alchemist51

.gitignore

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

sandbox/libs/composite-engine-lib/build.gradle

@@ -0,0 +1,27 @@
+/*
+ * 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.
+ */
+
+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-composite-engine-lib'
+  }
+}
+
+testingConventions.enabled = true
+
+tasks.named('forbiddenApisMain').configure {
+    replaceSignatureFiles 'jdk-signatures'
+}

sandbox/libs/composite-engine-lib/src/main/java/org/opensearch/composite/queue/ConcurrentQueue.java

@@ -0,0 +1,140 @@
+/*
+ * 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.composite.queue;
+
+import java.util.Iterator;
+import java.util.Queue;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Predicate;
+import java.util.function.Supplier;
+
+/**
+ * A striped concurrent queue that distributes entries across multiple internal
+ * queues using thread-affinity-based hashing. This reduces contention by allowing
+ * concurrent threads to operate on different stripes without blocking each other.
+ *
+ * @param <T> the type of elements held in this queue
+ * @opensearch.experimental
+ */
+public final class ConcurrentQueue<T> {
+
+    static final int MIN_CONCURRENCY = 1;
+    static final int MAX_CONCURRENCY = 256;
+
+    private final int concurrency;
+    private final Lock[] locks;
+    private final Queue<T>[] queues;
+    private final Supplier<Queue<T>> queueSupplier;
+
+    ConcurrentQueue(Supplier<Queue<T>> queueSupplier, int concurrency) {
+        if (concurrency < MIN_CONCURRENCY || concurrency > MAX_CONCURRENCY) {
+            throw new IllegalArgumentException(
+                "concurrency must be in [" + MIN_CONCURRENCY + ", " + MAX_CONCURRENCY + "], got " + concurrency
+            );
+        }
+        this.concurrency = concurrency;
+        this.queueSupplier = queueSupplier;
+        locks = new Lock[concurrency];
+        @SuppressWarnings({ "rawtypes", "unchecked" })
+        Queue<T>[] queues = new Queue[concurrency];
+        this.queues = queues;
+        for (int i = 0; i < concurrency; ++i) {
+            locks[i] = new ReentrantLock();
+            queues[i] = queueSupplier.get();
+        }
+    }
+
+    void add(T entry) {
+        // Seed the order in which to look at entries based on the current thread. This helps distribute
+        // entries across queues and gives a bit of thread affinity between entries and threads, which
+        // can't hurt.
+        final int threadHash = Thread.currentThread().hashCode() & 0xFFFF;
+        for (int i = 0; i < concurrency; ++i) {
+            final int index = (threadHash + i) % concurrency;
+            final Lock lock = locks[index];
+            final Queue<T> queue = queues[index];
+            if (lock.tryLock()) {
+                try {
+                    queue.add(entry);
+                    return;
+                } finally {
+                    lock.unlock();
+                }
+            }
+        }
+        final int index = threadHash % concurrency;
+        final Lock lock = locks[index];
+        final Queue<T> queue = queues[index];
+        lock.lock();
+        try {
+            queue.add(entry);
+        } finally {
+            lock.unlock();
+        }
+    }
+
+    T poll(Predicate<T> predicate) {
+        final int threadHash = Thread.currentThread().hashCode() & 0xFFFF;
+        for (int i = 0; i < concurrency; ++i) {
+            final int index = (threadHash + i) % concurrency;
+            final Lock lock = locks[index];
+            final Queue<T> queue = queues[index];
+            if (lock.tryLock()) {
+                try {
+                    Iterator<T> it = queue.iterator();
+                    while (it.hasNext()) {
+                        T entry = it.next();
+                        if (predicate.test(entry)) {
+                            it.remove();
+                            return entry;
+                        }
+                    }
+                } finally {
+                    lock.unlock();
+                }
+            }
+        }
+        for (int i = 0; i < concurrency; ++i) {
+            final int index = (threadHash + i) % concurrency;
+            final Lock lock = locks[index];
+            final Queue<T> queue = queues[index];
+            lock.lock();
+            try {
+                Iterator<T> it = queue.iterator();
+                while (it.hasNext()) {
+                    T entry = it.next();
+                    if (predicate.test(entry)) {
+                        it.remove();
+                        return entry;
+                    }
+                }
+            } finally {
+                lock.unlock();
+            }
+        }
+        return null;
+    }
+
+    boolean remove(T entry) {
+        for (int i = 0; i < concurrency; ++i) {
+            final Lock lock = locks[i];
+            final Queue<T> queue = queues[i];
+            lock.lock();
+            try {
+                if (queue.remove(entry)) {
+                    return true;
+                }
+            } finally {
+                lock.unlock();
+            }
+        }
+        return false;
+    }
+}

sandbox/libs/composite-engine-lib/src/main/java/org/opensearch/composite/queue/LockableConcurrentQueue.java

@@ -0,0 +1,80 @@
+/*
+ * 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.composite.queue;
+
+import java.util.Queue;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.Lock;
+import java.util.function.Supplier;
+
+/**
+ * A concurrent queue wrapper that adds lock-and-poll / add-and-unlock semantics
+ * on top of {@link ConcurrentQueue}. Entries must implement {@link Lock} so that
+ * they can be atomically locked when polled and unlocked when returned.
+ * <p>
+ * This is used by the composite writer pool to ensure that a writer is locked
+ * before it is handed out and unlocked when it is returned.
+ *
+ * @param <T> the type of lockable elements held in this queue
+ * @opensearch.experimental
+ */
+public final class LockableConcurrentQueue<T extends Lock> {
+
+    private final ConcurrentQueue<T> queue;
+    private final AtomicInteger addAndUnlockCounter = new AtomicInteger();
+
+    /**
+     * Creates a new lockable concurrent queue.
+     *
+     * @param queueSupplier supplier for the underlying queue instances
+     * @param concurrency   the concurrency level (number of stripes)
+     */
+    public LockableConcurrentQueue(Supplier<Queue<T>> queueSupplier, int concurrency) {
+        this.queue = new ConcurrentQueue<>(queueSupplier, concurrency);
+    }
+
+    /**
+     * Lock an entry, and poll it from the queue, in that order. If no entry can be found and locked,
+     * {@code null} is returned.
+     */
+    public T lockAndPoll() {
+        int addAndUnlockCount;
+        do {
+            addAndUnlockCount = addAndUnlockCounter.get();
+            T entry = queue.poll(Lock::tryLock);
+            if (entry != null) {
+                return entry;
+            }
+            // If an entry has been added to the queue in the meantime, try again.
+        } while (addAndUnlockCount != addAndUnlockCounter.get());
+
+        return null;
+    }
+
+    /**
+     * Remove an entry from the queue.
+     *
+     * @param entry the entry to remove
+     * @return {@code true} if the entry was removed
+     */
+    public boolean remove(T entry) {
+        return queue.remove(entry);
+    }
+
+    /**
+     * Add an entry to the queue and unlock it, in that order.
+     *
+     * @param entry the entry to add and unlock
+     */
+    public void addAndUnlock(T entry) {
+        queue.add(entry);
+        entry.unlock();
+        addAndUnlockCounter.incrementAndGet();
+    }
+}

sandbox/libs/composite-engine-lib/src/main/java/org/opensearch/composite/queue/package-info.java

@@ -0,0 +1,14 @@
+/*
+ * 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.
+ */
+
+/**
+ * Concurrent queue utilities for the composite indexing engine.
+ *
+ * @opensearch.experimental
+ */
+package org.opensearch.composite.queue;