Add realtime segment update and announcement to replace action

indexing-service/src/main/java/org/apache/druid/indexing/common/actions/SegmentTransactionalReplaceAction.java

@@ -136,6 +136,12 @@ public SegmentPublishResult perform(Task task, TaskActionToolbox toolbox)
             upgradedPendingSegments.size(), task.getId(), upgradedPendingSegments
         );
 
+        for (String supervisorId : activeSupervisorIds) {
+          for (SegmentIdWithShardSpec pendingSegment : upgradedPendingSegments) {
+            toolbox.getSupervisorManager().updatePendingSegmentMapping(supervisorId, pendingSegment);
+          }
+        }
+
         // These upgraded pending segments should be forwarded to the SupervisorManager
       }
       catch (Exception e) {

indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskClientAsyncImpl.java

@@ -37,7 +37,6 @@
 import org.apache.druid.indexing.common.TaskInfoProvider;
 import org.apache.druid.java.util.common.Either;
 import org.apache.druid.java.util.common.ISE;
-import org.apache.druid.java.util.common.Pair;
 import org.apache.druid.java.util.common.StringUtils;
 import org.apache.druid.java.util.common.concurrent.Execs;
 import org.apache.druid.java.util.common.jackson.JacksonUtils;
@@ -203,10 +202,16 @@ public ListenableFuture<Boolean> updatePendingSegmentMapping(
       Set<SegmentIdWithShardSpec> versionsOfPendingSegment
   )
   {
+    if (versionsOfPendingSegment.isEmpty()) {
+      return Futures.immediateFuture(true);
+    }
+    final List<SegmentIdWithShardSpec> allVersionsOfPendingSegment = new ArrayList<>();
+    allVersionsOfPendingSegment.add(rootPendingSegment);
+    allVersionsOfPendingSegment.addAll(versionsOfPendingSegment);
     final RequestBuilder requestBuilder = new RequestBuilder(
         HttpMethod.POST,
-        "pendingSegmentMapping"
-    ).jsonContent(jsonMapper, Pair.of(rootPendingSegment, versionsOfPendingSegment));
+        "/pendingSegmentMapping"
+    ).jsonContent(jsonMapper, allVersionsOfPendingSegment);
 
     return makeRequest(id, requestBuilder)
         .handler(IgnoreHttpResponseHandler.INSTANCE)

indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java

@@ -1558,13 +1558,13 @@ public Response setEndOffsetsHTTP(
   @Consumes(MediaType.APPLICATION_JSON)
   @Produces(MediaType.APPLICATION_JSON)
   public Response updatePendingSegmentMapping(
-      Pair<SegmentIdWithShardSpec, Set<SegmentIdWithShardSpec>> rootPendingSegmentToVersionMapping,
+      List<SegmentIdWithShardSpec> allVersionsOfPendingSegment,
       // this field is only for internal purposes, shouldn't be usually set by users
       @Context final HttpServletRequest req
   )
   {
     authorizationCheck(req, Action.WRITE);
-    return updatePendingSegmentMapping(rootPendingSegmentToVersionMapping.lhs, rootPendingSegmentToVersionMapping.rhs);
+    return updatePendingSegmentMapping(allVersionsOfPendingSegment);
   }
 
   public Map<String, Object> doGetRowStats()
@@ -1772,13 +1772,13 @@ public Response setEndOffsets(
     return Response.ok(sequenceNumbers).build();
   }
 
-  private Response updatePendingSegmentMapping(
-      SegmentIdWithShardSpec rootPendingSegment,
-      Set<SegmentIdWithShardSpec> versionsOfPendingSegment
-  )
+  private Response updatePendingSegmentMapping(List<SegmentIdWithShardSpec> allVersionsOfPendingSegment)
   {
     try {
-      ((StreamAppenderator) appenderator).updatePendingSegmentMapping(rootPendingSegment, versionsOfPendingSegment);
+      ((StreamAppenderator) appenderator).updatePendingSegmentMapping(
+          allVersionsOfPendingSegment.get(0),
+          allVersionsOfPendingSegment.subList(1, allVersionsOfPendingSegment.size())
+      );
       return Response.ok().build();
     }
     catch (IOException e) {

indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java

@@ -1101,7 +1101,7 @@ public void updatePendingSegmentMapping(SegmentIdWithShardSpec rootPendingSegmen
         taskClient.updatePendingSegmentMapping(
             taskId,
             rootPendingSegment,
-            indexerMetadataStorageCoordinator.getAllVersionsOfPendingSegment(rootPendingSegment)
+            indexerMetadataStorageCoordinator.findAllVersionsOfPendingSegment(rootPendingSegment)
         );
       }
     }
@@ -1111,7 +1111,7 @@ public void updatePendingSegmentMapping(SegmentIdWithShardSpec rootPendingSegmen
           taskClient.updatePendingSegmentMapping(
               taskId,
               rootPendingSegment,
-              indexerMetadataStorageCoordinator.getAllVersionsOfPendingSegment(rootPendingSegment)
+              indexerMetadataStorageCoordinator.findAllVersionsOfPendingSegment(rootPendingSegment)
           );
         }
       }

server/src/main/java/org/apache/druid/indexing/overlord/IndexerMetadataStorageCoordinator.java

@@ -432,7 +432,4 @@ SegmentPublishResult commitMetadataOnly(
    * @return DataSegment used segment corresponding to given id
    */
   DataSegment retrieveSegmentForId(String id, boolean includeUnused);
-
-  Set<SegmentIdWithShardSpec> getAllVersionsOfPendingSegment(SegmentIdWithShardSpec rootPendingSegment);
-
 }

server/src/main/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinator.java

@@ -2554,12 +2554,6 @@ public DataSegment retrieveSegmentForId(final String id, boolean includeUnused)
     );
   }
 
-  @Override
-  public Set<SegmentIdWithShardSpec> getAllVersionsOfPendingSegment(SegmentIdWithShardSpec rootPendingSegment)
-  {
-    return Collections.emptySet();
-  }
-
   private static class PendingSegmentsRecord
   {
     private final String sequenceName;

server/src/main/java/org/apache/druid/segment/realtime/appenderator/SinkQuerySegmentWalker.java

@@ -69,10 +69,11 @@
 import org.joda.time.Interval;
 
 import java.io.Closeable;
-import java.util.Map;
+import java.util.List;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Function;
 
@@ -95,7 +96,8 @@ public class SinkQuerySegmentWalker implements QuerySegmentWalker
   private final Cache cache;
   private final CacheConfig cacheConfig;
   private final CachePopulatorStats cachePopulatorStats;
-  private final Map<SegmentDescriptor, SegmentDescriptor> newIdToRootPendingSegment = new ConcurrentHashMap<>();
+  private final ConcurrentMap<SegmentDescriptor, SegmentDescriptor> newIdToRootPendingSegment
+      = new ConcurrentHashMap<>();
 
   public SinkQuerySegmentWalker(
       String dataSource,
@@ -304,7 +306,7 @@ public <T> QueryRunner<T> getQueryRunnerForSegments(final Query<T> query, final
 
   public void updatePendingSegmentMapping(
       SegmentIdWithShardSpec rootPendingSegment,
-      Set<SegmentIdWithShardSpec> versionsOfPendingSegment
+      List<SegmentIdWithShardSpec> versionsOfPendingSegment
   )
   {
     for (SegmentIdWithShardSpec versionOfPendingSegment : versionsOfPendingSegment) {

server/src/main/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderator.java

@@ -1057,7 +1057,7 @@ private void unannounceRootSegmentAndUpgradedVersions(Sink sink) throws IOExcept
 
   public void updatePendingSegmentMapping(
       SegmentIdWithShardSpec rootPendingSegment,
-      Set<SegmentIdWithShardSpec> versionsOfPendingSegment
+      List<SegmentIdWithShardSpec> versionsOfPendingSegment
   ) throws IOException
   {
     if (!sinks.containsKey(rootPendingSegment) || droppingSinks.contains(rootPendingSegment)) {