Add CompactionMode to compaction Eligibility (split from #18968 by @cecemei) (#19054)

* Add CompactionMode to compaction Eligibility

* Deprecate old stuff

* Clean up comment

* Javadocs update

* Fix test

* revert job queue change

* Eligibility

---------

Co-authored-by: cecemei <yingqian.mei@gmail.com>

Author: kfaraz

indexing-service/src/main/java/org/apache/druid/indexing/compact/CompactionJobQueue.java

@@ -282,7 +282,7 @@ private boolean startJobIfPendingAndReady(
     }
 
     // Check if the job is already running, completed or skipped
-    final CompactionStatus compactionStatus = getCurrentStatusForJob(job, policy);
+    final CompactionStatus compactionStatus = statusTracker.computeCompactionStatus(job.getCandidate(), policy);
     switch (compactionStatus.getState()) {
       case RUNNING:
         return false;
@@ -378,14 +378,6 @@ private void persistPendingIndexingState(CompactionJob job)
     }
   }
 
-  public CompactionStatus getCurrentStatusForJob(CompactionJob job, CompactionCandidateSearchPolicy policy)
-  {
-    final CompactionStatus compactionStatus = statusTracker.computeCompactionStatus(job.getCandidate(), policy);
-    final CompactionCandidate candidatesWithStatus = job.getCandidate().withCurrentStatus(null);
-    statusTracker.onCompactionStatusComputed(candidatesWithStatus, null);
-    return compactionStatus;
-  }
-
   public static CompactionConfigValidationResult validateCompactionJob(BatchIndexingJob job)
   {
     // For MSQ jobs, do not perform any validation

server/src/main/java/org/apache/druid/server/compaction/CompactionCandidateSearchPolicy.java

@@ -24,6 +24,7 @@
 import org.apache.druid.java.util.common.StringUtils;
 import org.apache.druid.server.coordinator.duty.CompactSegments;
 
+import javax.annotation.Nullable;
 import java.util.Objects;
 
 /**
@@ -64,15 +65,18 @@ Eligibility checkEligibilityForCompaction(
    */
   class Eligibility
   {
-    public static final Eligibility OK = new Eligibility(true, null);
+    public static final Eligibility OK = new Eligibility(true, null, CompactionMode.FULL_COMPACTION);
 
     private final boolean eligible;
     private final String reason;
+    @Nullable
+    private final CompactionMode mode;
 
-    private Eligibility(boolean eligible, String reason)
+    private Eligibility(boolean eligible, String reason, @Nullable CompactionMode mode)
     {
       this.eligible = eligible;
       this.reason = reason;
+      this.mode = mode;
     }
 
     public boolean isEligible()
@@ -85,9 +89,19 @@ public String getReason()
       return reason;
     }
 
+    /**
+     * The mode of compaction (full or minor). This is non-null only when the
+     * candidate is considered to be eligible for compaction by the policy.
+     */
+    @Nullable
+    public CompactionMode getMode()
+    {
+      return mode;
+    }
+
     public static Eligibility fail(String messageFormat, Object... args)
     {
-      return new Eligibility(false, StringUtils.format(messageFormat, args));
+      return new Eligibility(false, StringUtils.format(messageFormat, args), null);
     }
 
     @Override
@@ -100,21 +114,22 @@ public boolean equals(Object object)
         return false;
       }
       Eligibility that = (Eligibility) object;
-      return eligible == that.eligible && Objects.equals(reason, that.reason);
+      return eligible == that.eligible && Objects.equals(reason, that.reason) && Objects.equals(mode, that.mode);
     }
 
     @Override
     public int hashCode()
     {
-      return Objects.hash(eligible, reason);
+      return Objects.hash(eligible, reason, mode);
     }
 
     @Override
     public String toString()
     {
       return "Eligibility{" +
              "eligible=" + eligible +
-             ", reason='" + reason + '\'' +
+             ", reason='" + reason +
+             ", mode='" + mode + '\'' +
              '}';
     }
   }

server/src/main/java/org/apache/druid/server/compaction/CompactionMode.java

@@ -0,0 +1,31 @@
+/*
+ * 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.compaction;
+
+/**
+ * Represents the mode of compaction for segment intervals.
+ */
+public enum CompactionMode
+{
+  /**
+   * Indicates that all existing segments of the interval will be picked for compaction.
+   */
+  FULL_COMPACTION;
+}

server/src/main/java/org/apache/druid/server/compaction/CompactionRunSimulator.java

@@ -49,7 +49,12 @@
 /**
  * Simulates runs of auto-compaction duty to obtain the expected list of
  * compaction tasks that would be submitted by the actual compaction duty.
+ *
+ * @deprecated The simulator does not support the Overlord-based CompactionJobQueue
+ * or the new reindexing templates. It will either be fully replaced or undergo
+ * a major overhaul in the upcoming releases.
  */
+@Deprecated
 public class CompactionRunSimulator
 {
   private final CompactionStatusTracker statusTracker;

server/src/main/java/org/apache/druid/server/compaction/CompactionStatusTracker.java

@@ -23,6 +23,7 @@
 import org.apache.druid.indexer.TaskStatus;
 import org.apache.druid.java.util.common.DateTimes;
 import org.apache.druid.server.coordinator.DataSourceCompactionConfig;
+import org.checkerframework.checker.nullness.qual.Nullable;
 import org.joda.time.DateTime;
 import org.joda.time.Duration;
 import org.joda.time.Interval;
@@ -58,6 +59,7 @@ public void removeDatasource(String datasource)
     datasourceStatuses.remove(datasource);
   }
 
+  @Nullable
   public CompactionTaskStatus getLatestTaskStatus(CompactionCandidate candidates)
   {
     return datasourceStatuses
@@ -79,7 +81,11 @@ public Set<String> getSubmittedTaskIds()
    * Checks if compaction can be started for the given {@link CompactionCandidate}.
    * This method assumes that the given candidate is eligible for compaction
    * based on the current compaction config/supervisor of the datasource.
+   *
+   * @deprecated This method is used only by Coordinator-based CompactSegments
+   * duty and will be removed in the future.
    */
+  @Deprecated
   public CompactionStatus computeCompactionStatus(
       CompactionCandidate candidate,
       CompactionCandidateSearchPolicy searchPolicy