Do not kill segments with referenced load specs from deep storage

indexing-service/src/main/java/org/apache/druid/indexing/appenderator/ActionBasedPublishedSegmentRetriever.java

@@ -79,7 +79,7 @@ public Set<DataSegment> findPublishedSegments(Set<SegmentId> segmentIds) throws
     catch (Exception e) {
       log.warn(
           e,
-          "Could not retrieve published segment IDs[%s] using task action[segmentListById]."
+          "Could not retrieve published segment IDs[%s] using task action[retrieveSegmentsById]."
           + " Overlord maybe on an older version, retrying with action[segmentListUsed]."
           + " This task may fail to publish segments if there is a concurrent replace happening.",
           serializedSegmentIds

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

@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.indexing.common.actions;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.core.type.TypeReference;
+import org.apache.druid.indexing.common.task.Task;
+
+import java.util.Set;
+
+/**
+ * Task action to retrieve the segment IDs from which a given set of segments were upgraded.
+ */
+public class RetrieveUpgradedFromSegmentIdsAction implements TaskAction<UpgradedFromSegmentsResponse>
+{
+  private final String dataSource;
+  private final Set<String> segmentIds;
+
+  @JsonCreator
+  public RetrieveUpgradedFromSegmentIdsAction(
+      @JsonProperty("dataSource") String dataSource,
+      @JsonProperty("segmentIds") Set<String> segmentIds
+  )
+  {
+    this.dataSource = dataSource;
+    this.segmentIds = segmentIds;
+  }
+
+  @JsonProperty
+  public String getDataSource()
+  {
+    return dataSource;
+  }
+
+  @JsonProperty
+  public Set<String> getSegmentIds()
+  {
+    return segmentIds;
+  }
+
+  @Override
+  public TypeReference<UpgradedFromSegmentsResponse> getReturnTypeReference()
+  {
+    return new TypeReference<UpgradedFromSegmentsResponse>()
+    {
+    };
+  }
+
+  @Override
+  public UpgradedFromSegmentsResponse perform(Task task, TaskActionToolbox toolbox)
+  {
+    return new UpgradedFromSegmentsResponse(
+        toolbox.getIndexerMetadataStorageCoordinator()
+               .retrieveUpgradedFromSegmentIds(dataSource, segmentIds)
+    );
+  }
+
+  @Override
+  public boolean isAudited()
+  {
+    return false;
+  }
+
+  @Override
+  public String toString()
+  {
+    return getClass().getSimpleName() + "{" +
+           "dataSource='" + dataSource + '\'' +
+           ", segmentIds=" + segmentIds +
+           '}';
+  }
+}

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

@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.indexing.common.actions;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.core.type.TypeReference;
+import org.apache.druid.indexing.common.task.Task;
+
+import java.util.Set;
+
+/**
+ * Task action to determine the set of all segments containing the same load spec given the parent id. <br/>
+ * Returns a map from a segment ID to a set containing:
+ * <ol>
+ * <li> all segment IDs that were upgraded from it AND are still present in the metadata store </li>
+ * <li> the segment ID itself if and only if it is still present in the metadata store </li>
+ * </ol>
+ */
+public class RetrieveUpgradedToSegmentIdsAction implements TaskAction<UpgradedToSegmentsResponse>
+{
+  private final String dataSource;
+  private final Set<String> segmentIds;
+
+  @JsonCreator
+  public RetrieveUpgradedToSegmentIdsAction(
+      @JsonProperty("dataSource") String dataSource,
+      @JsonProperty("segmentIds") Set<String> segmentIds
+  )
+  {
+    this.dataSource = dataSource;
+    this.segmentIds = segmentIds;
+  }
+
+  @JsonProperty
+  public String getDataSource()
+  {
+    return dataSource;
+  }
+
+  @JsonProperty
+  public Set<String> getSegmentIds()
+  {
+    return segmentIds;
+  }
+
+  @Override
+  public TypeReference<UpgradedToSegmentsResponse> getReturnTypeReference()
+  {
+    return new TypeReference<UpgradedToSegmentsResponse>()
+    {
+    };
+  }
+
+  @Override
+  public UpgradedToSegmentsResponse perform(Task task, TaskActionToolbox toolbox)
+  {
+    return new UpgradedToSegmentsResponse(
+        toolbox.getIndexerMetadataStorageCoordinator()
+               .retrieveUpgradedToSegmentIds(dataSource, segmentIds)
+    );
+  }
+
+  @Override
+  public boolean isAudited()
+  {
+    return false;
+  }
+
+  @Override
+  public String toString()
+  {
+    return getClass().getSimpleName() + "{" +
+           "dataSource='" + dataSource + '\'' +
+           ", segmentIds=" + segmentIds +
+           '}';
+  }
+}

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

@@ -39,6 +39,8 @@
     @JsonSubTypes.Type(name = "segmentTransactionalAppend", value = SegmentTransactionalAppendAction.class),
     @JsonSubTypes.Type(name = "segmentTransactionalReplace", value = SegmentTransactionalReplaceAction.class),
     @JsonSubTypes.Type(name = "retrieveSegmentsById", value = RetrieveSegmentsByIdAction.class),
+    @JsonSubTypes.Type(name = "retrieveUpgradedFromSegmentIds", value = RetrieveUpgradedFromSegmentIdsAction.class),
+    @JsonSubTypes.Type(name = "retrieveUpgradedToSegmentIds", value = RetrieveUpgradedToSegmentIdsAction.class),
     @JsonSubTypes.Type(name = "segmentListUsed", value = RetrieveUsedSegmentsAction.class),
     @JsonSubTypes.Type(name = "segmentListUnused", value = RetrieveUnusedSegmentsAction.class),
     @JsonSubTypes.Type(name = "markSegmentsAsUnused", value = MarkSegmentsAsUnusedAction.class),

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

@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.indexing.common.actions;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.util.Map;
+
+public class UpgradedFromSegmentsResponse
+{
+  private final Map<String, String> upgradedFromSegmentIds;
+
+  @JsonCreator
+  public UpgradedFromSegmentsResponse(
+      @JsonProperty("upgradedFromSegmentIds") Map<String, String> upgradedFromSegmentIds
+  )
+  {
+    this.upgradedFromSegmentIds = upgradedFromSegmentIds;
+  }
+
+  @JsonProperty
+  public Map<String, String> getUpgradedFromSegmentIds()
+  {
+    return upgradedFromSegmentIds;
+  }
+}

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

@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.indexing.common.actions;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.util.Map;
+import java.util.Set;
+
+public class UpgradedToSegmentsResponse
+{
+
+  private final Map<String, Set<String>> upgradedToSegmentIds;
+
+  @JsonCreator
+  public UpgradedToSegmentsResponse(
+      @JsonProperty("upgradedToSegmentIds") Map<String, Set<String>> upgradedToSegmentIds
+  )
+  {
+    this.upgradedToSegmentIds = upgradedToSegmentIds;
+  }
+
+  @JsonProperty
+  public Map<String, Set<String>> getUpgradedToSegmentIds()
+  {
+    return upgradedToSegmentIds;
+  }
+}