Add API to update compaction engine (apache#16803)

Changes:
- Add API `/druid/coordinator/v1/config/compaction/global` to update cluster level compaction config
- Add class `CompactionConfigUpdateRequest`
- Fix bug in `CoordinatorCompactionConfig` which caused compaction engine to not be persisted.
Use json field name `engine` instead of `compactionEngine` because JSON field names must align
with the getter name.
- Update MSQ validation error messages
- Complete overhaul of `CoordinatorCompactionConfigResourceTest` to remove unnecessary mocking
and add more meaningful tests.
- Add `TuningConfigBuilder` to easily build tuning configs for tests.
- Add `DatasourceCompactionConfigBuilder`

benchmarks/src/test/java/org/apache/druid/server/coordinator/NewestSegmentFirstPolicyBenchmark.java

@@ -92,21 +92,12 @@ public void setup()
       final String dataSource = DATA_SOURCE_PREFIX + i;
       compactionConfigs.put(
           dataSource,
-          new DataSourceCompactionConfig(
-              dataSource,
-              0,
-              inputSegmentSizeBytes,
-              null,
-              null,
-              null,
-              null,
-              null,
-              null,
-              null,
-              null,
-              null,
-              null
-          )
+          DataSourceCompactionConfig
+              .builder()
+              .forDataSource(dataSource)
+              .withTaskPriority(0)
+              .withInputSegmentSizeBytes(inputSegmentSizeBytes)
+              .build()
       );
     }
 

extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/MSQCompactionRunner.java

@@ -143,7 +143,7 @@ public CompactionConfigValidationResult validateCompactionTask(
     return validationResults.stream()
                             .filter(result -> !result.isValid())
                             .findFirst()
-                            .orElse(new CompactionConfigValidationResult(true, null));
+                            .orElse(CompactionConfigValidationResult.success());
   }
 
   @Override

extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/indexing/MSQCompactionRunnerTest.java

@@ -211,7 +211,7 @@ public void testMSQEngineWithUnsupportedMetricsSpecIsInValid()
     CompactionConfigValidationResult validationResult = MSQ_COMPACTION_RUNNER.validateCompactionTask(compactionTask);
     Assert.assertFalse(validationResult.isValid());
     Assert.assertEquals(
-        "Different name[sum_added] and fieldName(s)[[added]] for aggregator unsupported for MSQ engine.",
+        "MSQ: Different name[sum_added] and fieldName(s)[[added]] for aggregator",
         validationResult.getReason()
     );
   }

indexing-service/src/main/java/org/apache/druid/indexing/common/task/NativeCompactionRunner.java

@@ -88,7 +88,7 @@ public CompactionConfigValidationResult validateCompactionTask(
       CompactionTask compactionTask
   )
   {
-    return new CompactionConfigValidationResult(true, null);
+    return CompactionConfigValidationResult.success();
   }
 
   /**

server/src/main/java/org/apache/druid/client/indexing/ClientCompactionRunnerInfo.java

@@ -92,7 +92,7 @@ public static CompactionConfigValidationResult validateCompactionConfig(
   {
     CompactionEngine compactionEngine = newConfig.getEngine() == null ? defaultCompactionEngine : newConfig.getEngine();
     if (compactionEngine == CompactionEngine.NATIVE) {
-      return new CompactionConfigValidationResult(true, null);
+      return CompactionConfigValidationResult.success();
     } else {
       return compactionConfigSupportedByMSQEngine(newConfig);
     }
@@ -125,7 +125,7 @@ private static CompactionConfigValidationResult compactionConfigSupportedByMSQEn
     return validationResults.stream()
                             .filter(result -> !result.isValid())
                             .findFirst()
-                            .orElse(new CompactionConfigValidationResult(true, null));
+                            .orElse(CompactionConfigValidationResult.success());
   }
 
   /**
@@ -135,22 +135,19 @@ public static CompactionConfigValidationResult validatePartitionsSpecForMSQ(Part
   {
     if (!(partitionsSpec instanceof DimensionRangePartitionsSpec
           || partitionsSpec instanceof DynamicPartitionsSpec)) {
-      return new CompactionConfigValidationResult(
-          false,
-          "Invalid partitionsSpec type[%s] for MSQ engine. Type must be either 'dynamic' or 'range'.",
+      return CompactionConfigValidationResult.failure(
+          "MSQ: Invalid partitioning type[%s]. Must be either 'dynamic' or 'range'",
           partitionsSpec.getClass().getSimpleName()
 
       );
     }
     if (partitionsSpec instanceof DynamicPartitionsSpec
         && ((DynamicPartitionsSpec) partitionsSpec).getMaxTotalRows() != null) {
-      return new CompactionConfigValidationResult(
-          false,
-          "maxTotalRows[%d] in DynamicPartitionsSpec not supported for MSQ engine.",
-          ((DynamicPartitionsSpec) partitionsSpec).getMaxTotalRows()
+      return CompactionConfigValidationResult.failure(
+          "MSQ: 'maxTotalRows' not supported with 'dynamic' partitioning"
       );
     }
-    return new CompactionConfigValidationResult(true, null);
+    return CompactionConfigValidationResult.success();
   }
 
   /**
@@ -162,12 +159,11 @@ public static CompactionConfigValidationResult validateRollupForMSQ(
   )
   {
     if (metricsSpec != null && isRollup != null && !isRollup) {
-      return new CompactionConfigValidationResult(
-          false,
-          "rollup in granularitySpec must be set to True if metricsSpec is specifed for MSQ engine."
+      return CompactionConfigValidationResult.failure(
+          "MSQ: 'granularitySpec.rollup' must be true if 'metricsSpec' is specified"
       );
     }
-    return new CompactionConfigValidationResult(true, null);
+    return CompactionConfigValidationResult.success();
   }
 
   /**
@@ -179,14 +175,13 @@ public static CompactionConfigValidationResult validateMaxNumTasksForMSQ(Map<Str
       int maxNumTasks = QueryContext.of(context)
                                     .getInt(ClientMSQContext.CTX_MAX_NUM_TASKS, ClientMSQContext.DEFAULT_MAX_NUM_TASKS);
       if (maxNumTasks < 2) {
-        return new CompactionConfigValidationResult(false,
-                                                    "MSQ context maxNumTasks [%,d] cannot be less than 2, "
-                                                    + "since at least 1 controller and 1 worker is necessary.",
-                                                    maxNumTasks
+        return CompactionConfigValidationResult.failure(
+            "MSQ: Context maxNumTasks[%,d] must be at least 2 (1 controller + 1 worker)",
+            maxNumTasks
         );
       }
     }
-    return new CompactionConfigValidationResult(true, null);
+    return CompactionConfigValidationResult.success();
   }
 
   /**
@@ -195,7 +190,7 @@ public static CompactionConfigValidationResult validateMaxNumTasksForMSQ(Map<Str
   public static CompactionConfigValidationResult validateMetricsSpecForMSQ(AggregatorFactory[] metricsSpec)
   {
     if (metricsSpec == null) {
-      return new CompactionConfigValidationResult(true, null);
+      return CompactionConfigValidationResult.success();
     }
     return Arrays.stream(metricsSpec)
                  .filter(aggregatorFactory ->
@@ -206,11 +201,10 @@ public static CompactionConfigValidationResult validateMetricsSpecForMSQ(Aggrega
                                                       .equals(aggregatorFactory.getName())))
                  .findFirst()
                  .map(aggregatorFactory ->
-                          new CompactionConfigValidationResult(
-                              false,
-                              "Different name[%s] and fieldName(s)[%s] for aggregator unsupported for MSQ engine.",
+                          CompactionConfigValidationResult.failure(
+                              "MSQ: Different name[%s] and fieldName(s)[%s] for aggregator",
                               aggregatorFactory.getName(),
                               aggregatorFactory.requiredFields()
-                          )).orElse(new CompactionConfigValidationResult(true, null));
+                          )).orElse(CompactionConfigValidationResult.success());
   }
 }

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

@@ -23,10 +23,23 @@
 
 public class CompactionConfigValidationResult
 {
+  private static final CompactionConfigValidationResult SUCCESS
+      = new CompactionConfigValidationResult(true, null);
+
   private final boolean valid;
   private final String reason;
 
-  public CompactionConfigValidationResult(boolean valid, String format, Object... args)
+  public static CompactionConfigValidationResult success()
+  {
+    return SUCCESS;
+  }
+
+  public static CompactionConfigValidationResult failure(String msgFormat, Object... args)
+  {
+    return new CompactionConfigValidationResult(false, msgFormat, args);
+  }
+
+  private CompactionConfigValidationResult(boolean valid, String format, Object... args)
   {
     this.valid = valid;
     this.reason = format == null ? null : StringUtils.format(format, args);

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

@@ -24,6 +24,7 @@
 import com.google.common.collect.ImmutableList;
 import org.apache.druid.common.config.Configs;
 import org.apache.druid.indexer.CompactionEngine;
+import org.apache.druid.server.http.CompactionConfigUpdateRequest;
 
 import javax.annotation.Nullable;
 import java.util.List;
@@ -54,23 +55,21 @@ public static CoordinatorCompactionConfig from(
         baseConfig.compactionTaskSlotRatio,
         baseConfig.maxCompactionTaskSlots,
         baseConfig.useAutoScaleSlots,
-        null
+        baseConfig.compactionEngine
     );
   }
 
   public static CoordinatorCompactionConfig from(
       CoordinatorCompactionConfig baseConfig,
-      @Nullable Double compactionTaskSlotRatio,
-      @Nullable Integer maxCompactionTaskSlots,
-      @Nullable Boolean useAutoScaleSlots
+      CompactionConfigUpdateRequest update
   )
   {
     return new CoordinatorCompactionConfig(
         baseConfig.compactionConfigs,
-        compactionTaskSlotRatio == null ? baseConfig.compactionTaskSlotRatio : compactionTaskSlotRatio,
-        maxCompactionTaskSlots == null ? baseConfig.maxCompactionTaskSlots : maxCompactionTaskSlots,
-        useAutoScaleSlots == null ? baseConfig.useAutoScaleSlots : useAutoScaleSlots,
-        null
+        Configs.valueOrDefault(update.getCompactionTaskSlotRatio(), baseConfig.compactionTaskSlotRatio),
+        Configs.valueOrDefault(update.getMaxCompactionTaskSlots(), baseConfig.maxCompactionTaskSlots),
+        Configs.valueOrDefault(update.getUseAutoScaleSlots(), baseConfig.useAutoScaleSlots),
+        Configs.valueOrDefault(update.getCompactionEngine(), baseConfig.compactionEngine)
     );
   }
 
@@ -90,7 +89,7 @@ public CoordinatorCompactionConfig(
       @JsonProperty("compactionTaskSlotRatio") @Nullable Double compactionTaskSlotRatio,
       @JsonProperty("maxCompactionTaskSlots") @Nullable Integer maxCompactionTaskSlots,
       @JsonProperty("useAutoScaleSlots") @Nullable Boolean useAutoScaleSlots,
-      @JsonProperty("compactionEngine") @Nullable CompactionEngine compactionEngine
+      @JsonProperty("engine") @Nullable CompactionEngine compactionEngine
   )
   {
     this.compactionConfigs = compactionConfigs;

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

@@ -24,6 +24,7 @@
 import com.google.common.base.Preconditions;
 import org.apache.druid.indexer.CompactionEngine;
 import org.apache.druid.query.aggregation.AggregatorFactory;
+import org.apache.druid.server.coordinator.config.DataSourceCompactionConfigBuilder;
 import org.joda.time.Period;
 
 import javax.annotation.Nullable;
@@ -42,6 +43,12 @@ public class DataSourceCompactionConfig
   private final String dataSource;
   private final int taskPriority;
   private final long inputSegmentSizeBytes;
+
+  public static DataSourceCompactionConfigBuilder builder()
+  {
+    return new DataSourceCompactionConfigBuilder();
+  }
+
   /**
    * The number of input segments is limited because the byte size of a serialized task spec is limited by
    * org.apache.druid.indexing.overlord.config.RemoteTaskRunnerConfig.maxZnodeBytes.