Invalidate cache load status on historical as soon as it is served

indexing-service/src/test/java/org/apache/druid/indexing/common/task/AppenderatorDriverRealtimeIndexTaskTest.java

@@ -72,7 +72,6 @@
 import org.apache.druid.indexing.overlord.TaskLockbox;
 import org.apache.druid.indexing.overlord.TaskStorage;
 import org.apache.druid.indexing.overlord.supervisor.SupervisorManager;
-import org.apache.druid.indexing.test.TestDataSegmentAnnouncer;
 import org.apache.druid.indexing.test.TestDataSegmentKiller;
 import org.apache.druid.indexing.test.TestDataSegmentPusher;
 import org.apache.druid.jackson.DefaultObjectMapper;
@@ -125,6 +124,7 @@
 import org.apache.druid.server.DruidNode;
 import org.apache.druid.server.coordination.DataSegmentServerAnnouncer;
 import org.apache.druid.server.coordination.ServerType;
+import org.apache.druid.server.coordination.TestDataSegmentAnnouncer;
 import org.apache.druid.server.security.AuthTestUtils;
 import org.apache.druid.testing.InitializedNullHandlingTest;
 import org.apache.druid.timeline.DataSegment;

indexing-service/src/test/java/org/apache/druid/indexing/common/task/RealtimeIndexTaskTest.java

@@ -59,7 +59,6 @@
 import org.apache.druid.indexing.overlord.TaskLockbox;
 import org.apache.druid.indexing.overlord.TaskStorage;
 import org.apache.druid.indexing.overlord.supervisor.SupervisorManager;
-import org.apache.druid.indexing.test.TestDataSegmentAnnouncer;
 import org.apache.druid.indexing.test.TestDataSegmentKiller;
 import org.apache.druid.indexing.test.TestDataSegmentPusher;
 import org.apache.druid.indexing.test.TestIndexerMetadataStorageCoordinator;
@@ -115,6 +114,7 @@
 import org.apache.druid.server.DruidNode;
 import org.apache.druid.server.coordination.DataSegmentServerAnnouncer;
 import org.apache.druid.server.coordination.ServerType;
+import org.apache.druid.server.coordination.TestDataSegmentAnnouncer;
 import org.apache.druid.server.security.AuthTestUtils;
 import org.apache.druid.testing.InitializedNullHandlingTest;
 import org.apache.druid.timeline.DataSegment;

indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskTestBase.java

@@ -75,7 +75,6 @@
 import org.apache.druid.indexing.overlord.TaskLockbox;
 import org.apache.druid.indexing.overlord.TaskStorage;
 import org.apache.druid.indexing.overlord.supervisor.SupervisorManager;
-import org.apache.druid.indexing.test.TestDataSegmentAnnouncer;
 import org.apache.druid.indexing.test.TestDataSegmentKiller;
 import org.apache.druid.java.util.common.ISE;
 import org.apache.druid.java.util.common.Intervals;
@@ -120,6 +119,7 @@
 import org.apache.druid.server.DruidNode;
 import org.apache.druid.server.coordination.DataSegmentServerAnnouncer;
 import org.apache.druid.server.coordination.ServerType;
+import org.apache.druid.server.coordination.TestDataSegmentAnnouncer;
 import org.apache.druid.server.security.AuthTestUtils;
 import org.apache.druid.timeline.DataSegment;
 import org.apache.druid.utils.CompressionUtils;

server/src/main/java/org/apache/druid/client/HttpServerInventoryView.java

@@ -608,14 +608,13 @@ public void deltaSync(List<DataSegmentChangeRequest> changes)
         {
           for (DataSegmentChangeRequest request : changes) {
             if (request instanceof SegmentChangeRequestLoad) {
-              addSegment(((SegmentChangeRequestLoad) request).getSegment(), false);
+              addSegment(request.getSegment(), false);
             } else if (request instanceof SegmentChangeRequestDrop) {
-              removeSegment(((SegmentChangeRequestDrop) request).getSegment(), false);
+              removeSegment(request.getSegment(), false);
             } else {
               log.error(
                   "Server[%s] gave a non load/drop dataSegmentChangeRequest[%s], Ignored.",
-                  druidServer.getName(),
-                  request
+                  druidServer.getName(), request
               );
             }
           }

server/src/main/java/org/apache/druid/server/coordination/DataSegmentChangeRequest.java

@@ -21,6 +21,7 @@
 
 import com.fasterxml.jackson.annotation.JsonSubTypes;
 import com.fasterxml.jackson.annotation.JsonTypeInfo;
+import org.apache.druid.timeline.DataSegment;
 
 import javax.annotation.Nullable;
 
@@ -37,4 +38,6 @@ public interface DataSegmentChangeRequest
   void go(DataSegmentChangeHandler handler, @Nullable DataSegmentChangeCallback callback);
 
   String asString();
+
+  DataSegment getSegment();
 }

server/src/main/java/org/apache/druid/server/coordination/DataSegmentChangeResponse.java

@@ -0,0 +1,137 @@
+/*
+ * 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.server.coordination;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.base.Preconditions;
+
+import javax.annotation.Nullable;
+
+/**
+ * Response of a {@link DataSegmentChangeRequest}. Contains the request itself
+ * and the {@link Status} of the request.
+ */
+public class DataSegmentChangeResponse
+{
+  private final DataSegmentChangeRequest request;
+  private final Status status;
+
+  @JsonCreator
+  public DataSegmentChangeResponse(
+      @JsonProperty("request") DataSegmentChangeRequest request,
+      @JsonProperty("status") Status status
+  )
+  {
+    this.request = request;
+    this.status = status;
+  }
+
+  @JsonProperty
+  public DataSegmentChangeRequest getRequest()
+  {
+    return request;
+  }
+
+  @JsonProperty
+  public Status getStatus()
+  {
+    return status;
+  }
+
+  @JsonIgnore
+  public boolean isComplete()
+  {
+    return getStatus().getState() != State.PENDING;
+  }
+
+  @JsonIgnore
+  public boolean isLoadRequest()
+  {
+    return request instanceof SegmentChangeRequestLoad;
+  }
+
+  @Override
+  public String toString()
+  {
+    return "DataSegmentChangeResponse{" +
+           "request=" + request +
+           ", status=" + status +
+           '}';
+  }
+
+  public enum State
+  {
+    SUCCESS, FAILED, PENDING
+  }
+
+  /**
+   * Contains {@link State} of a {@link DataSegmentChangeRequest} and the failure
+   * message, if any.
+   */
+  public static class Status
+  {
+    private final State state;
+    @Nullable
+    private final String failureCause;
+
+    public static final Status SUCCESS = new Status(State.SUCCESS, null);
+    public static final Status PENDING = new Status(State.PENDING, null);
+
+    @JsonCreator
+    public Status(
+        @JsonProperty("state") State state,
+        @JsonProperty("failureCause") @Nullable String failureCause
+    )
+    {
+      Preconditions.checkNotNull(state, "state must be non-null");
+      this.state = state;
+      this.failureCause = failureCause;
+    }
+
+    public static Status failed(String cause)
+    {
+      return new Status(State.FAILED, cause);
+    }
+
+    @JsonProperty
+    public State getState()
+    {
+      return state;
+    }
+
+    @Nullable
+    @JsonProperty
+    public String getFailureCause()
+    {
+      return failureCause;
+    }
+
+    @Override
+    public String toString()
+    {
+      return "Status{" +
+             "state=" + state +
+             ", failureCause='" + failureCause + '\'' +
+             '}';
+    }
+  }
+}

server/src/main/java/org/apache/druid/server/coordination/SegmentChangeRequestDrop.java

@@ -45,6 +45,7 @@ public SegmentChangeRequestDrop(
 
   @JsonProperty
   @JsonUnwrapped
+  @Override
   public DataSegment getSegment()
   {
     return segment;

server/src/main/java/org/apache/druid/server/coordination/SegmentChangeRequestLoad.java

@@ -63,6 +63,7 @@ public void go(DataSegmentChangeHandler handler, @Nullable DataSegmentChangeCall
 
   @JsonProperty
   @JsonUnwrapped
+  @Override
   public DataSegment getSegment()
   {
     return segment;

server/src/main/java/org/apache/druid/server/coordination/SegmentChangeRequestNoop.java

@@ -19,6 +19,9 @@
 
 package org.apache.druid.server.coordination;
 
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import org.apache.druid.timeline.DataSegment;
+
 import javax.annotation.Nullable;
 
 /**
@@ -39,4 +42,11 @@ public String asString()
   {
     return "NOOP";
   }
+
+  @Override
+  @JsonIgnore
+  public DataSegment getSegment()
+  {
+    throw new UnsupportedOperationException();
+  }
 }