Create RFS successor work items

RFS/src/main/java/org/opensearch/migrations/bulkload/tracing/IWorkCoordinationContexts.java

@@ -13,6 +13,7 @@ class ActivityNames {
         public static final String ACQUIRE_SPECIFIC_WORK = "acquireSpecificWorkItem";
         public static final String COMPLETE_WORK = "completeWork";
         public static final String ACQUIRE_NEXT_WORK = "acquireNextWorkItem";
+        public static final String CREATE_SUCCESSOR_WORK_ITEMS = "createSuccessorWorkItems";
 
         private ActivityNames() {}
     }
@@ -54,6 +55,8 @@ interface IBaseAcquireWorkContext extends IRetryableActivityContext {}
 
     interface IAcquireSpecificWorkContext extends IBaseAcquireWorkContext {
         String ACTIVITY_NAME = ActivityNames.ACQUIRE_SPECIFIC_WORK;
+
+        ICreateSuccessorWorkItemsContext getCreateSuccessorWorkItemsContext();
     }
 
     interface IAcquireNextWorkItemContext extends IBaseAcquireWorkContext {
@@ -68,6 +71,9 @@ interface IAcquireNextWorkItemContext extends IBaseAcquireWorkContext {
         void recordRecoverableClockError();
 
         void recordFailure(OpenSearchWorkCoordinator.PotentialClockDriftDetectedException e);
+
+        ICreateSuccessorWorkItemsContext getCreateSuccessorWorkItemsContext();
+
     }
 
     interface ICompleteWorkItemContext extends IRetryableActivityContext {
@@ -76,6 +82,13 @@ interface ICompleteWorkItemContext extends IRetryableActivityContext {
         IRefreshContext getRefreshContext();
     }
 
+    interface ICreateSuccessorWorkItemsContext extends IRetryableActivityContext {
+        String ACTIVITY_NAME = ActivityNames.CREATE_SUCCESSOR_WORK_ITEMS;
+        IRefreshContext getRefreshContext();
+        ICompleteWorkItemContext getCompleteWorkItemContext();
+        ICreateUnassignedWorkItemContext getCreateUnassignedWorkItemContext();
+    }
+
     interface IScopedWorkContext<C extends IBaseAcquireWorkContext> extends IScopedInstrumentationAttributes {
         C createOpeningContext();
 

RFS/src/main/java/org/opensearch/migrations/bulkload/tracing/RootWorkCoordinationContext.java

@@ -18,6 +18,7 @@ public class RootWorkCoordinationContext extends RootOtelContext {
     public final WorkCoordinationContexts.AcquireSpecificWorkContext.MetricInstruments acquireSpecificWorkMetrics;
     public final WorkCoordinationContexts.CompleteWorkItemContext.MetricInstruments completeWorkMetrics;
     public final WorkCoordinationContexts.AcquireNextWorkItemContext.MetricInstruments acquireNextWorkMetrics;
+    public final WorkCoordinationContexts.CreateSuccessorWorkItemsContext.MetricInstruments createSuccessorWorkItemsMetrics;
 
     public RootWorkCoordinationContext(OpenTelemetry sdk, IContextTracker contextTracker) {
         this(sdk, contextTracker, null);
@@ -38,6 +39,7 @@ public RootWorkCoordinationContext(OpenTelemetry sdk,
         acquireSpecificWorkMetrics = WorkCoordinationContexts.AcquireSpecificWorkContext.makeMetrics(meter);
         completeWorkMetrics = WorkCoordinationContexts.CompleteWorkItemContext.makeMetrics(meter);
         acquireNextWorkMetrics = WorkCoordinationContexts.AcquireNextWorkItemContext.makeMetrics(meter);
+        createSuccessorWorkItemsMetrics = WorkCoordinationContexts.CreateSuccessorWorkItemsContext.makeMetrics(meter);
     }
 
     public IWorkCoordinationContexts.IInitializeCoordinatorStateContext createCoordinationInitializationStateContext() {
@@ -87,4 +89,13 @@ public IWorkCoordinationContexts.ICompleteWorkItemContext createCompleteWorkCont
     ) {
         return new WorkCoordinationContexts.CompleteWorkItemContext(this, enclosingScope);
     }
-}
+
+    public IWorkCoordinationContexts.ICreateSuccessorWorkItemsContext createSuccessorWorkItemsContext() {
+        return createSuccessorWorkItemsContext(null);
+    }
+
+    public IWorkCoordinationContexts.ICreateSuccessorWorkItemsContext createSuccessorWorkItemsContext(
+            IScopedInstrumentationAttributes enclosingScope
+    ) {
+        return new WorkCoordinationContexts.CreateSuccessorWorkItemsContext(this, enclosingScope);
+    }}

RFS/src/main/java/org/opensearch/migrations/bulkload/tracing/WorkCoordinationContexts.java

@@ -248,6 +248,10 @@ private MetricInstruments(Meter meter, String activityName) {
         public MetricInstruments getRetryMetrics() {
             return getRootInstrumentationScope().acquireSpecificWorkMetrics;
         }
+
+        public ICreateSuccessorWorkItemsContext  getCreateSuccessorWorkItemsContext() {
+            return new CreateSuccessorWorkItemsContext(this.rootInstrumentationScope, this);
+        }
     }
 
     @Getter
@@ -276,6 +280,10 @@ public IRefreshContext getRefreshContext() {
             return new Refresh(this.rootInstrumentationScope, this);
         }
 
+        public ICreateSuccessorWorkItemsContext  getCreateSuccessorWorkItemsContext() {
+            return new CreateSuccessorWorkItemsContext(this.rootInstrumentationScope, this);
+        }
+
         public static class MetricInstruments extends RetryMetricInstruments {
             public final LongCounter assignedCounter;
             public final LongCounter nothingAvailableCounter;
@@ -363,4 +371,51 @@ public MetricInstruments getRetryMetrics() {
             return getRootInstrumentationScope().completeWorkMetrics;
         }
     }
+
+    @Getter
+    class CreateSuccessorWorkItemsContext extends BaseSpanContext<RootWorkCoordinationContext>
+            implements
+            ICreateSuccessorWorkItemsContext,
+            RetryableActivityContextMetricMixin<CreateSuccessorWorkItemsContext.MetricInstruments> {
+        final IScopedInstrumentationAttributes enclosingScope;
+
+        CreateSuccessorWorkItemsContext(
+                RootWorkCoordinationContext rootScope,
+                IScopedInstrumentationAttributes enclosingScope
+        ) {
+            super(rootScope);
+            this.enclosingScope = enclosingScope;
+            initializeSpan(rootScope);
+        }
+
+        @Override
+        public String getActivityName() {
+            return ACTIVITY_NAME;
+        }
+
+        @Override
+        public IRefreshContext getRefreshContext() {
+            return new Refresh(this.rootInstrumentationScope, this);
+        }
+
+        @Override
+        public ICompleteWorkItemContext getCompleteWorkItemContext() {
+            return new CompleteWorkItemContext(this.rootInstrumentationScope, this);
+        }
+
+        public static class MetricInstruments extends RetryMetricInstruments {
+            private MetricInstruments(Meter meter, String activityName) {
+                super(meter, autoLabels(activityName));
+            }
+        }
+
+        public static @NonNull MetricInstruments makeMetrics(Meter meter) {
+            return new MetricInstruments(meter, ACTIVITY_NAME);
+        }
+
+        @Override
+        public MetricInstruments getRetryMetrics() {
+            return getRootInstrumentationScope().createSuccessorWorkItemsMetrics;
+        }
+    }
 }

RFS/src/main/java/org/opensearch/migrations/bulkload/workcoordination/IWorkCoordinator.java

@@ -4,6 +4,7 @@
 import java.time.Clock;
 import java.time.Duration;
 import java.time.Instant;
+import java.util.ArrayList;
 import java.util.function.Supplier;
 
 import org.opensearch.migrations.bulkload.tracing.IWorkCoordinationContexts;
@@ -67,8 +68,8 @@ WorkAcquisitionOutcome createOrUpdateLeaseForWorkItem(
         String workItemId,
         Duration leaseDuration,
         Supplier<IWorkCoordinationContexts.IAcquireSpecificWorkContext> contextSupplier
-    ) throws IOException;
-
+    ) throws IOException, InterruptedException;
+    
     /**
      * Scan the created work items that have not yet had leases acquired and have not yet finished.
      * One of those work items will be returned along with a lease for how long this process may continue
@@ -98,6 +99,20 @@ void completeWorkItem(
         Supplier<IWorkCoordinationContexts.ICompleteWorkItemContext> contextSupplier
     ) throws IOException, InterruptedException;
 
+    /**
+     * Add the list of successor items to the work item, create new work items for each of the successors, and mark the
+     * original work item as completed.
+     * @param workItemId the work item that is being completed
+     * @param successorWorkItemIds the list of successor work items that will be created
+     * @throws IOException
+     * @throws InterruptedException
+     */
+    void createSuccessorWorkItemsAndMarkComplete(
+        String workItemId,
+        ArrayList<String> successorWorkItemIds,
+        Supplier<IWorkCoordinationContexts.ICreateSuccessorWorkItemsContext> contextSupplier
+    ) throws IOException, InterruptedException;
+
     /**
      * @return the number of items that are not yet complete.  This will include items with and without claimed leases.
      * @throws IOException