Move writerPool implementation out of IndexingExecutionEngine

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

Author: alchemist51

…x/libs/composite-engine-lib/build.gradle libs/concurrent-queue/build.gradlesandbox/libs/composite-engine-lib/build.gradle renamed to libs/concurrent-queue/build.gradle sandbox/libs/composite-engine-lib/build.gradle renamed to libs/concurrent-queue/build.gradle

@@ -12,16 +12,22 @@
  */
 
 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-composite-engine-lib'
+    exclude group: 'org.opensearch', module: 'opensearch-concurrent-queue'
   }
 }
 
 testingConventions.enabled = true
 
 tasks.named('forbiddenApisMain').configure {
-    replaceSignatureFiles 'jdk-signatures'
+  // :libs:opensearch-concurrent-queue does not depend on server
+  replaceSignatureFiles 'jdk-signatures'
 }

…rch/composite/queue/ConcurrentQueue.java …search/common/queue/ConcurrentQueue.javasandbox/libs/composite-engine-lib/src/main/java/org/opensearch/composite/queue/ConcurrentQueue.java renamed to libs/concurrent-queue/src/main/java/org/opensearch/common/queue/ConcurrentQueue.java sandbox/libs/composite-engine-lib/src/main/java/org/opensearch/composite/queue/ConcurrentQueue.java renamed to libs/concurrent-queue/src/main/java/org/opensearch/common/queue/ConcurrentQueue.java

@@ -6,7 +6,7 @@
  * compatible open source license.
  */
 
-package org.opensearch.composite.queue;
+package org.opensearch.common.queue;
 
 import java.util.Iterator;
 import java.util.Queue;

…opensearch/composite/queue/Lockable.java …rg/opensearch/common/queue/Lockable.javasandbox/libs/composite-engine-lib/src/main/java/org/opensearch/composite/queue/Lockable.java renamed to libs/concurrent-queue/src/main/java/org/opensearch/common/queue/Lockable.java sandbox/libs/composite-engine-lib/src/main/java/org/opensearch/composite/queue/Lockable.java renamed to libs/concurrent-queue/src/main/java/org/opensearch/common/queue/Lockable.java

@@ -6,16 +6,13 @@
  * compatible open source license.
  */
 
-package org.opensearch.composite.queue;
-
-import jdk.jfr.Experimental;
+package org.opensearch.common.queue;
 
 /**
  * A minimal locking contract for objects managed by a {@link LockableConcurrentQueue}.
  *
  * @opensearch.experimental
  */
-@Experimental
 public interface Lockable {
 
     /**

…osite/queue/LockableConcurrentQueue.java …ommon/queue/LockableConcurrentQueue.javasandbox/libs/composite-engine-lib/src/main/java/org/opensearch/composite/queue/LockableConcurrentQueue.java renamed to libs/concurrent-queue/src/main/java/org/opensearch/common/queue/LockableConcurrentQueue.java sandbox/libs/composite-engine-lib/src/main/java/org/opensearch/composite/queue/LockableConcurrentQueue.java renamed to libs/concurrent-queue/src/main/java/org/opensearch/common/queue/LockableConcurrentQueue.java

@@ -6,7 +6,7 @@
  * compatible open source license.
  */
 
-package org.opensearch.composite.queue;
+package org.opensearch.common.queue;
 
 import java.util.Queue;
 import java.util.concurrent.atomic.AtomicInteger;

…search/composite/queue/package-info.java …pensearch/common/queue/package-info.javasandbox/libs/composite-engine-lib/src/main/java/org/opensearch/composite/queue/package-info.java renamed to libs/concurrent-queue/src/main/java/org/opensearch/common/queue/package-info.java sandbox/libs/composite-engine-lib/src/main/java/org/opensearch/composite/queue/package-info.java renamed to libs/concurrent-queue/src/main/java/org/opensearch/common/queue/package-info.java

@@ -11,4 +11,4 @@
  *
  * @opensearch.experimental
  */
-package org.opensearch.composite.queue;
+package org.opensearch.common.queue;

…omposite/queue/ConcurrentQueueTests.java …h/common/queue/ConcurrentQueueTests.javasandbox/libs/composite-engine-lib/src/test/java/org/opensearch/composite/queue/ConcurrentQueueTests.java renamed to libs/concurrent-queue/src/test/java/org/opensearch/common/queue/ConcurrentQueueTests.java sandbox/libs/composite-engine-lib/src/test/java/org/opensearch/composite/queue/ConcurrentQueueTests.java renamed to libs/concurrent-queue/src/test/java/org/opensearch/common/queue/ConcurrentQueueTests.java

@@ -6,7 +6,7 @@
  * compatible open source license.
  */
 
-package org.opensearch.composite.queue;
+package org.opensearch.common.queue;
 
 import org.opensearch.test.OpenSearchTestCase;
 

…/queue/LockableConcurrentQueueTests.java …/queue/LockableConcurrentQueueTests.javasandbox/libs/composite-engine-lib/src/test/java/org/opensearch/composite/queue/LockableConcurrentQueueTests.java renamed to libs/concurrent-queue/src/test/java/org/opensearch/common/queue/LockableConcurrentQueueTests.java sandbox/libs/composite-engine-lib/src/test/java/org/opensearch/composite/queue/LockableConcurrentQueueTests.java renamed to libs/concurrent-queue/src/test/java/org/opensearch/common/queue/LockableConcurrentQueueTests.java

@@ -6,7 +6,7 @@
  * compatible open source license.
  */
 
-package org.opensearch.composite.queue;
+package org.opensearch.common.queue;
 
 import org.opensearch.test.OpenSearchTestCase;
 

sandbox/plugins/composite-engine/build.gradle

@@ -12,7 +12,7 @@ opensearchplugin {
 }
 
 dependencies {
-  api project(':sandbox:libs:composite-engine-lib')
+  api project(':libs:opensearch-concurrent-queue')
   compileOnly project(':server')
   testImplementation project(':test:framework')
 }

sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/CompositeDataFormatWriterPool.java

@@ -10,13 +10,15 @@
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.opensearch.common.Nullable;
 import org.opensearch.common.annotation.ExperimentalApi;
 import org.opensearch.common.settings.Setting;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.index.IndexModule;
 import org.opensearch.index.IndexSettings;
 import org.opensearch.index.engine.dataformat.DataFormat;
 import org.opensearch.index.engine.dataformat.DataFormatPlugin;
+import org.opensearch.index.engine.dataformat.DataformatAwareLockableWriterPool;
 import org.opensearch.index.engine.dataformat.IndexingExecutionEngine;
 import org.opensearch.index.mapper.MapperService;
 import org.opensearch.index.shard.ShardPath;
@@ -28,6 +30,7 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 
 /**
  * Sandbox plugin that provides a {@link CompositeIndexingExecutionEngine} for
@@ -186,14 +189,14 @@ public DataFormat getDataFormat() {
     public IndexingExecutionEngine<?, ?> indexingEngine(
         MapperService mapperService,
         ShardPath shardPath,
-        IndexSettings indexSettings
+        IndexSettings indexSettings,
+        @Nullable DataformatAwareLockableWriterPool<?> writerPool
     ) {
-        return new CompositeIndexingExecutionEngine(
-            dataFormatPlugins,
-            indexSettings,
-            mapperService,
-            shardPath
-        );
+        Objects.requireNonNull(writerPool, "DataformatAwareLockableWriterPool is required for CompositeIndexingExecutionEngine");
+        @SuppressWarnings("unchecked")
+        DataformatAwareLockableWriterPool<CompositeWriter> compositeWriterPool = (DataformatAwareLockableWriterPool<
+            CompositeWriter>) writerPool;
+        return new CompositeIndexingExecutionEngine(dataFormatPlugins, indexSettings, mapperService, shardPath, compositeWriterPool);
     }
 
     /**