Add some fixes

Author: alchemist51

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

@@ -55,8 +55,8 @@ public class CompositeIndexingExecutionEngine implements IndexingExecutionEngine
 
     private static final Logger logger = LogManager.getLogger(CompositeIndexingExecutionEngine.class);
 
-    private final IndexingExecutionEngine<? extends DataFormat, ? extends DocumentInput<?>> primaryEngine;
-    private final List<IndexingExecutionEngine<? extends DataFormat, ? extends DocumentInput<?>>> secondaryEngines;
+    private final IndexingExecutionEngine<?, ?> primaryEngine;
+    private final List<IndexingExecutionEngine<?, ?>> secondaryEngines;
     private final CompositeDataFormat compositeDataFormat;
     private final CompositeDataFormatWriterPool dataFormatWriterPool;
     private final AtomicLong writerGenerationCounter;
@@ -278,7 +278,7 @@ public CompositeDocumentInput newDocumentInput() {
      *
      * @return the secondary engines
      */
-    public List<IndexingExecutionEngine<? extends DataFormat, ? extends DocumentInput<?>>> getSecondaryDelegates() {
+    public List<IndexingExecutionEngine<?, ?>> getSecondaryDelegates() {
         return secondaryEngines;
     }
 

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

@@ -46,8 +46,8 @@ public class CompositeWriter implements Writer<CompositeDocumentInput>, Lock {
 
     private static final Logger logger = LogManager.getLogger(CompositeWriter.class);
 
-    private final Map.Entry<DataFormat, Writer<? extends DocumentInput<?>>> primaryWriter;
-    private final List<Map.Entry<DataFormat, Writer<? extends DocumentInput<?>>>> secondaryWriters;
+    private final Map.Entry<DataFormat, Writer<DocumentInput<?>>> primaryWriter;
+    private final List<Map.Entry<DataFormat, Writer<DocumentInput<?>>>> secondaryWriters;
     private final ReentrantLock lock;
     private final long writerGeneration;
     private final RowIdGenerator rowIdGenerator;
@@ -64,6 +64,7 @@ public class CompositeWriter implements Writer<CompositeDocumentInput>, Lock {
      * @param engine           the composite indexing execution engine
      * @param writerGeneration the writer generation number
      */
+    @SuppressWarnings("unchecked")
     public CompositeWriter(CompositeIndexingExecutionEngine engine, long writerGeneration) {
         this.lock = new ReentrantLock();
         this.aborted = false;
@@ -73,22 +74,26 @@ public CompositeWriter(CompositeIndexingExecutionEngine engine, long writerGener
         IndexingExecutionEngine<?, ?> primaryDelegate = engine.getPrimaryDelegate();
         this.primaryWriter = new AbstractMap.SimpleImmutableEntry<>(
             primaryDelegate.getDataFormat(),
-            primaryDelegate.createWriter(writerGeneration)
+            (Writer<DocumentInput<?>>) primaryDelegate.createWriter(writerGeneration)
         );
 
-        List<Map.Entry<DataFormat, Writer<?>>> secondaries = new ArrayList<>();
+        List<Map.Entry<DataFormat, Writer<DocumentInput<?>>>> secondaries = new ArrayList<>();
         for (IndexingExecutionEngine<?, ?> delegate : engine.getSecondaryDelegates()) {
-            secondaries.add(new AbstractMap.SimpleImmutableEntry<>(delegate.getDataFormat(), delegate.createWriter(writerGeneration)));
+            secondaries.add(
+                new AbstractMap.SimpleImmutableEntry<>(
+                    delegate.getDataFormat(),
+                    (Writer<DocumentInput<?>>) delegate.createWriter(writerGeneration)
+                )
+            );
         }
         this.secondaryWriters = List.copyOf(secondaries);
         this.rowIdGenerator = new RowIdGenerator(CompositeWriter.class.getName());
     }
 
-    @SuppressWarnings("unchecked")
     @Override
     public WriteResult addDoc(CompositeDocumentInput doc) throws IOException {
         // Write to primary first
-        WriteResult primaryResult = ((Writer<DocumentInput<?>>) primaryWriter.getValue()).addDoc(doc.getPrimaryInput());
+        WriteResult primaryResult = primaryWriter.getValue().addDoc(doc.getPrimaryInput());
         switch (primaryResult) {
             case WriteResult.Success s -> logger.trace(
                 "Successfully added document in primary format [{}]",
@@ -103,9 +108,9 @@ public WriteResult addDoc(CompositeDocumentInput doc) throws IOException {
         // Then write to each secondary
         List<DocumentInput<?>> secondaryInputs = new ArrayList<>(doc.getSecondaryInputs().values());
         for (int i = 0; i < secondaryWriters.size(); i++) {
-            Map.Entry<DataFormat, Writer<?>> entry = secondaryWriters.get(i);
+            Map.Entry<DataFormat, Writer<DocumentInput<?>>> entry = secondaryWriters.get(i);
             DocumentInput<?> input = secondaryInputs.get(i);
-            WriteResult result = ((Writer<DocumentInput<?>>) entry.getValue()).addDoc(input);
+            WriteResult result = entry.getValue().addDoc(input);
             switch (result) {
                 case WriteResult.Success s -> logger.trace(
                     "Successfully added document in secondary format [{}]",
@@ -128,7 +133,7 @@ public FileInfos flush() throws IOException {
         Optional<WriterFileSet> primaryWfs = primaryWriter.getValue().flush().getWriterFileSet(primaryWriter.getKey());
         primaryWfs.ifPresent(writerFileSet -> builder.putWriterFileSet(primaryWriter.getKey(), writerFileSet));
         // Flush secondaries
-        for (Map.Entry<DataFormat, Writer<?>> entry : secondaryWriters) {
+        for (Map.Entry<DataFormat, Writer<DocumentInput<?>>> entry : secondaryWriters) {
             Optional<WriterFileSet> wfs = entry.getValue().flush().getWriterFileSet(entry.getKey());
             wfs.ifPresent(writerFileSet -> builder.putWriterFileSet(entry.getKey(), writerFileSet));
         }
@@ -138,7 +143,7 @@ public FileInfos flush() throws IOException {
     @Override
     public void sync() throws IOException {
         primaryWriter.getValue().sync();
-        for (Map.Entry<DataFormat, Writer<?>> entry : secondaryWriters) {
+        for (Map.Entry<DataFormat, Writer<DocumentInput<?>>> entry : secondaryWriters) {
             entry.getValue().sync();
         }
     }
@@ -150,7 +155,7 @@ public void close() {
         } catch (Exception e) {
             logger.warn("Failed to close primary Writer for format [{}]", primaryWriter.getKey().name(), e);
         }
-        for (Map.Entry<DataFormat, Writer<?>> entry : secondaryWriters) {
+        for (Map.Entry<DataFormat, Writer<DocumentInput<?>>> entry : secondaryWriters) {
             try {
                 entry.getValue().close();
             } catch (Exception e) {