Allow MSQ engine only for compaction supervisors

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

@@ -25,6 +25,7 @@
 import org.apache.druid.indexing.overlord.supervisor.SupervisorStateManager;
 import org.apache.druid.indexing.overlord.supervisor.autoscaler.LagStats;
 import org.apache.druid.java.util.common.DateTimes;
+import org.apache.druid.java.util.common.StringUtils;
 import org.apache.druid.java.util.common.logger.Logger;
 import org.apache.druid.server.coordinator.AutoCompactionSnapshot;
 
@@ -55,6 +56,13 @@ public void start()
     if (supervisorSpec.isSuspended()) {
       log.info("Suspending compaction for dataSource[%s].", dataSource);
       scheduler.stopCompaction(dataSource);
+    } else if (!supervisorSpec.getValidationResult().isValid()) {
+      log.warn(
+          "Cannot start compaction supervisor for datasource[%s] since the compaction supervisor spec is invalid. "
+          + "Reason[%s].",
+          dataSource,
+          supervisorSpec.getValidationResult().getReason()
+      );
     } else {
       log.info("Starting compaction for dataSource[%s].", dataSource);
       scheduler.startCompaction(dataSource, supervisorSpec.getSpec());
@@ -76,6 +84,13 @@ public SupervisorReport<AutoCompactionSnapshot> getStatus()
       snapshot = AutoCompactionSnapshot.builder(dataSource)
                                        .withStatus(AutoCompactionSnapshot.AutoCompactionScheduleStatus.NOT_ENABLED)
                                        .build();
+    } else if (!supervisorSpec.getValidationResult().isValid()) {
+      snapshot = AutoCompactionSnapshot.builder(dataSource)
+                                       .withMessage(StringUtils.format(
+                                           "Compaction supervisor spec is invalid. Reason[%s].",
+                                           supervisorSpec.getValidationResult().getReason()
+                                       ))
+                                       .build();
     } else {
       snapshot = scheduler.getCompactionSnapshot(dataSource);
     }
@@ -90,6 +105,8 @@ public SupervisorStateManager.State getState()
       return State.SCHEDULER_STOPPED;
     } else if (supervisorSpec.isSuspended()) {
       return State.SUSPENDED;
+    } else if (!supervisorSpec.getValidationResult().isValid()) {
+      return State.INVALID_SPEC;
     } else {
       return State.RUNNING;
     }
@@ -132,6 +149,7 @@ public enum State implements SupervisorStateManager.State
     SCHEDULER_STOPPED(true),
     RUNNING(true),
     SUSPENDED(true),
+    INVALID_SPEC(false),
     UNHEALTHY(false);
 
     private final boolean healthy;

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

@@ -23,7 +23,6 @@
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import org.apache.druid.common.config.Configs;
-import org.apache.druid.error.InvalidInput;
 import org.apache.druid.indexing.overlord.supervisor.SupervisorSpec;
 import org.apache.druid.server.coordinator.CompactionConfigValidationResult;
 import org.apache.druid.server.coordinator.DataSourceCompactionConfig;
@@ -40,6 +39,7 @@ public class CompactionSupervisorSpec implements SupervisorSpec
   private final boolean suspended;
   private final DataSourceCompactionConfig spec;
   private final CompactionScheduler scheduler;
+  private final CompactionConfigValidationResult validationResult;
 
   @JsonCreator
   public CompactionSupervisorSpec(
@@ -48,14 +48,10 @@ public CompactionSupervisorSpec(
       @JacksonInject CompactionScheduler scheduler
   )
   {
-    final CompactionConfigValidationResult validationResult = scheduler.validateCompactionConfig(spec);
-    if (!validationResult.isValid()) {
-      throw InvalidInput.exception("Compaction supervisor 'spec' is invalid. Reason[%s].", validationResult.getReason());
-    }
-
     this.spec = spec;
     this.suspended = Configs.valueOrDefault(suspended, false);
     this.scheduler = scheduler;
+    this.validationResult = scheduler.validateCompactionConfig(spec);
   }
 
   @JsonProperty
@@ -77,6 +73,11 @@ public String getId()
     return ID_PREFIX + spec.getDataSource();
   }
 
+  public CompactionConfigValidationResult getValidationResult()
+  {
+    return validationResult;
+  }
+
   @Override
   public CompactionSupervisor createSupervisor()
   {

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

@@ -202,7 +202,7 @@ public CompactionConfigValidationResult validateCompactionConfig(DataSourceCompa
     } else {
       return ClientCompactionRunnerInfo.validateCompactionConfig(
           compactionConfig,
-          compactionConfigSupplier.get().getEngine()
+          supervisorConfig.getEngine()
       );
     }
   }
@@ -272,7 +272,7 @@ private synchronized void scheduledRun()
   private synchronized void runCompactionDuty()
   {
     final CoordinatorRunStats stats = new CoordinatorRunStats();
-    duty.run(getLatestConfig(), getCurrentDatasourceTimelines(), stats);
+    duty.run(getLatestConfig(), getCurrentDatasourceTimelines(), supervisorConfig.getEngine(), stats);
 
     // Emit stats only if emission period has elapsed
     if (!sinceStatsEmitted.isRunning() || sinceStatsEmitted.hasElapsed(METRIC_EMISSION_PERIOD)) {
@@ -309,7 +309,8 @@ public CompactionSimulateResult simulateRunWithConfigUpdate(ClusterCompactionCon
     if (isRunning()) {
       return new CompactionRunSimulator(statusTracker, overlordClient).simulateRunWithConfig(
           getLatestConfig().withClusterConfig(updateRequest),
-          getCurrentDatasourceTimelines()
+          getCurrentDatasourceTimelines(),
+          supervisorConfig.getEngine()
       );
     } else {
       return new CompactionSimulateResult(Collections.emptyMap());

indexing-service/src/test/java/org/apache/druid/indexing/compact/CompactionSupervisorSpecTest.java

@@ -82,20 +82,29 @@ public void testSerdeOfSuspendedSpec()
   }
 
   @Test
-  public void testInvalidSpecThrowsException()
+  public void testGetStatusWithInvalidSpec()
   {
     Mockito.when(scheduler.validateCompactionConfig(ArgumentMatchers.any()))
            .thenReturn(CompactionConfigValidationResult.failure("bad spec"));
-    final DruidException exception = Assert.assertThrows(
-        DruidException.class,
-        () -> new CompactionSupervisorSpec(null, false, scheduler)
-    );
     Assert.assertEquals(
-        "Compaction supervisor 'spec' is invalid. Reason[bad spec].",
-        exception.getMessage()
+        "Compaction supervisor spec is invalid. Reason[bad spec].", new CompactionSupervisorSpec(
+            new DataSourceCompactionConfig.Builder().forDataSource("datasource").build(),
+            false,
+            scheduler
+        ).createSupervisor().getStatus().getPayload().getMessage()
     );
   }
 
+  @Test
+  public void testGetValidationResultForInvalidSpec()
+  {
+    Mockito.when(scheduler.validateCompactionConfig(ArgumentMatchers.any()))
+           .thenReturn(CompactionConfigValidationResult.failure("bad spec"));
+    CompactionConfigValidationResult validationResult = new CompactionSupervisorSpec(null, false, scheduler).getValidationResult();
+    Assert.assertFalse(validationResult.isValid());
+    Assert.assertEquals("bad spec", validationResult.getReason());
+  }
+
   @Test
   public void testGetIdAndDataSources()
   {

indexing-service/src/test/java/org/apache/druid/indexing/compact/OverlordCompactionSchedulerTest.java

@@ -120,7 +120,7 @@ public void setUp()
     serviceEmitter = new StubServiceEmitter();
     segmentsMetadataManager = new TestSegmentsMetadataManager();
 
-    supervisorConfig = new CompactionSupervisorConfig(true);
+    supervisorConfig = new CompactionSupervisorConfig(true, null);
     compactionConfig = DruidCompactionConfig.empty();
     coordinatorOverlordServiceConfig = new CoordinatorOverlordServiceConfig(false, null);
 
@@ -149,7 +149,7 @@ private void initScheduler()
   @Test
   public void testStartStopWhenSchedulerIsEnabled()
   {
-    supervisorConfig = new CompactionSupervisorConfig(true);
+    supervisorConfig = new CompactionSupervisorConfig(true, null);
     Assert.assertFalse(scheduler.isRunning());
 
     scheduler.start();
@@ -168,7 +168,7 @@ public void testStartStopWhenSchedulerIsEnabled()
   @Test
   public void testStartStopWhenScheduledIsDisabled()
   {
-    supervisorConfig = new CompactionSupervisorConfig(false);
+    supervisorConfig = new CompactionSupervisorConfig(false, null);
     initScheduler();
 
     Assert.assertFalse(scheduler.isRunning());
@@ -183,7 +183,7 @@ public void testStartStopWhenScheduledIsDisabled()
   @Test
   public void testSegmentsAreNotPolledWhenSchedulerIsDisabled()
   {
-    supervisorConfig = new CompactionSupervisorConfig(false);
+    supervisorConfig = new CompactionSupervisorConfig(false, null);
     initScheduler();
 
     verifySegmentPolling(false);
@@ -337,7 +337,7 @@ public void testRunSimulation()
     );
 
     final CompactionSimulateResult simulateResult = scheduler.simulateRunWithConfigUpdate(
-        new ClusterCompactionConfig(null, null, null, null, null)
+        new ClusterCompactionConfig(null, null, null, null)
     );
     Assert.assertEquals(1, simulateResult.getCompactionStates().size());
     final Table pendingCompactionTable = simulateResult.getCompactionStates().get(CompactionStatus.State.PENDING);
@@ -362,7 +362,7 @@ public void testRunSimulation()
     scheduler.stopCompaction(TestDataSource.WIKI);
 
     final CompactionSimulateResult simulateResultWhenDisabled = scheduler.simulateRunWithConfigUpdate(
-        new ClusterCompactionConfig(null, null, null, null, null)
+        new ClusterCompactionConfig(null, null, null, null)
     );
     Assert.assertTrue(simulateResultWhenDisabled.getCompactionStates().isEmpty());
 

indexing-service/src/test/java/org/apache/druid/indexing/overlord/http/OverlordCompactionResourceTest.java

@@ -44,9 +44,9 @@
 public class OverlordCompactionResourceTest
 {
   private static final CompactionSupervisorConfig SUPERVISOR_ENABLED
-      = new CompactionSupervisorConfig(true);
+      = new CompactionSupervisorConfig(true, null);
   private static final CompactionSupervisorConfig SUPERVISOR_DISABLED
-      = new CompactionSupervisorConfig(false);
+      = new CompactionSupervisorConfig(false, null);
 
   private CompactionScheduler scheduler;
 

integration-tests/src/test/java/org/apache/druid/tests/coordinator/duty/ITAutoCompactionTest.java

@@ -118,7 +118,7 @@ public class ITAutoCompactionTest extends AbstractIndexerTest
   @DataProvider(name = "engine")
   public static Object[][] engine()
   {
-    return new Object[][]{{CompactionEngine.NATIVE}, {CompactionEngine.MSQ}};
+    return new Object[][]{{CompactionEngine.NATIVE}};
   }
 
   @Inject

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

@@ -27,6 +27,7 @@
 import org.apache.druid.client.indexing.IndexingWorkerInfo;
 import org.apache.druid.client.indexing.TaskPayloadResponse;
 import org.apache.druid.client.indexing.TaskStatusResponse;
+import org.apache.druid.indexer.CompactionEngine;
 import org.apache.druid.indexer.TaskStatus;
 import org.apache.druid.indexer.TaskStatusPlus;
 import org.apache.druid.indexer.report.TaskReport;
@@ -75,7 +76,8 @@ public CompactionRunSimulator(
    */
   public CompactionSimulateResult simulateRunWithConfig(
       DruidCompactionConfig compactionConfig,
-      Map<String, SegmentTimeline> datasourceTimelines
+      Map<String, SegmentTimeline> datasourceTimelines,
+      CompactionEngine defaultEngine
   )
   {
     final Table compactedIntervals
@@ -138,13 +140,14 @@ public void onTaskSubmitted(ClientCompactionTaskQuery taskPayload, CompactionCan
 
     // Unlimited task slots to ensure that simulator does not skip any interval
     final DruidCompactionConfig configWithUnlimitedTaskSlots = compactionConfig.withClusterConfig(
-        new ClusterCompactionConfig(1.0, Integer.MAX_VALUE, null, null, null)
+        new ClusterCompactionConfig(1.0, Integer.MAX_VALUE, null, null)
     );
 
     final CoordinatorRunStats stats = new CoordinatorRunStats();
     new CompactSegments(simulationStatusTracker, readOnlyOverlordClient).run(
         configWithUnlimitedTaskSlots,
         datasourceTimelines,
+        defaultEngine,
         stats
     );
 

server/src/main/java/org/apache/druid/server/coordinator/AutoCompactionSnapshot.java

@@ -25,6 +25,7 @@
 import org.apache.druid.java.util.common.ISE;
 import org.apache.druid.server.compaction.CompactionStatistics;
 
+import javax.annotation.Nullable;
 import javax.validation.constraints.NotNull;
 import java.util.Objects;
 
@@ -41,6 +42,8 @@ public enum AutoCompactionScheduleStatus
   @JsonProperty
   private final AutoCompactionScheduleStatus scheduleStatus;
   @JsonProperty
+  private final String message;
+  @JsonProperty
   private final long bytesAwaitingCompaction;
   @JsonProperty
   private final long bytesCompacted;
@@ -68,6 +71,7 @@ public static Builder builder(String dataSource)
   public AutoCompactionSnapshot(
       @JsonProperty("dataSource") @NotNull String dataSource,
       @JsonProperty("scheduleStatus") @NotNull AutoCompactionScheduleStatus scheduleStatus,
+      @JsonProperty("message") @Nullable String message,
       @JsonProperty("bytesAwaitingCompaction") long bytesAwaitingCompaction,
       @JsonProperty("bytesCompacted") long bytesCompacted,
       @JsonProperty("bytesSkipped") long bytesSkipped,
@@ -81,6 +85,7 @@ public AutoCompactionSnapshot(
   {
     this.dataSource = dataSource;
     this.scheduleStatus = scheduleStatus;
+    this.message = message;
     this.bytesAwaitingCompaction = bytesAwaitingCompaction;
     this.bytesCompacted = bytesCompacted;
     this.bytesSkipped = bytesSkipped;
@@ -104,6 +109,12 @@ public AutoCompactionScheduleStatus getScheduleStatus()
     return scheduleStatus;
   }
 
+  @Nullable
+  public String getMessage()
+  {
+    return message;
+  }
+
   public long getBytesAwaitingCompaction()
   {
     return bytesAwaitingCompaction;
@@ -169,7 +180,8 @@ public boolean equals(Object o)
            intervalCountCompacted == that.intervalCountCompacted &&
            intervalCountSkipped == that.intervalCountSkipped &&
            dataSource.equals(that.dataSource) &&
-           scheduleStatus == that.scheduleStatus;
+           scheduleStatus == that.scheduleStatus &&
+           Objects.equals(message, that.message);
   }
 
   @Override
@@ -178,6 +190,7 @@ public int hashCode()
     return Objects.hash(
         dataSource,
         scheduleStatus,
+        message,
         bytesAwaitingCompaction,
         bytesCompacted,
         bytesSkipped,
@@ -194,6 +207,7 @@ public static class Builder
   {
     private final String dataSource;
     private AutoCompactionScheduleStatus scheduleStatus;
+    private String message;
 
     private final CompactionStatistics compactedStats = new CompactionStatistics();
     private final CompactionStatistics skippedStats = new CompactionStatistics();
@@ -215,6 +229,12 @@ public Builder withStatus(AutoCompactionScheduleStatus status)
       return this;
     }
 
+    public Builder withMessage(String message)
+    {
+      this.message = message;
+      return this;
+    }
+
     public void incrementWaitingStats(CompactionStatistics entry)
     {
       waitingStats.increment(entry);
@@ -235,6 +255,7 @@ public AutoCompactionSnapshot build()
       return new AutoCompactionSnapshot(
           dataSource,
           scheduleStatus,
+          message,
           waitingStats.getTotalBytes(),
           compactedStats.getTotalBytes(),
           skippedStats.getTotalBytes(),