You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by kf...@apache.org on 2025/03/23 02:59:54 UTC
(druid) branch master updated: Use compaction dynamic config to enable compaction supervisors (#17782)
This is an automated email from the ASF dual-hosted git repository.
kfaraz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new 5fa69bd285b Use compaction dynamic config to enable compaction supervisors (#17782)
5fa69bd285b is described below
commit 5fa69bd285bf3e5699c6debe20ba89eefffe4fec
Author: Kashif Faraz <ka...@gmail.com>
AuthorDate: Sun Mar 23 08:29:49 2025 +0530
Use compaction dynamic config to enable compaction supervisors (#17782)
Changes
---------
- Remove runtime property object `CompactionSupervisorConfig`
- Add fields `useSupervisors` and `engine` to cluster-level compaction dynamic config
- Remove unused field `useAutoScaleSlots`
---
docs/api-reference/automatic-compaction-api.md | 111 ++++++++++++++++++++-
docs/data-management/automatic-compaction.md | 9 +-
.../org/apache/druid/guice/SupervisorModule.java | 2 -
.../indexing/compact/CompactionScheduler.java | 8 +-
.../compact/OverlordCompactionScheduler.java | 17 ++--
.../overlord/http/OverlordCompactionResource.java | 10 +-
.../compact/OverlordCompactionSchedulerTest.java | 27 ++---
.../http/OverlordCompactionResourceTest.java | 31 +++---
.../testsEx/utils/DruidClusterAdminClient.java | 2 +-
.../clients/CompactionResourceTestClient.java | 21 ++--
.../testing/utils/DruidClusterAdminClient.java | 2 +-
.../duty/ITAutoCompactionLockContentionTest.java | 2 +-
.../coordinator/duty/ITAutoCompactionTest.java | 37 ++-----
.../server/compaction/CompactionRunSimulator.java | 2 +-
.../coordinator/ClusterCompactionConfig.java | 72 +++++++++----
.../coordinator/CompactionSupervisorConfig.java | 96 ------------------
.../server/coordinator/DruidCompactionConfig.java | 108 ++++++++++----------
.../http/CoordinatorCompactionConfigsResource.java | 4 +-
.../CompactionSupervisorConfigTest.java | 60 -----------
.../DataSourceCompactionConfigAuditEntryTest.java | 12 +--
.../DataSourceCompactionConfigHistoryTest.java | 2 +-
.../coordinator/DruidCompactionConfigTest.java | 17 +++-
.../server/coordinator/DruidCoordinatorTest.java | 2 +-
.../coordinator/duty/CompactSegmentsTest.java | 17 +---
.../CoordinatorCompactionConfigsResourceTest.java | 11 +-
25 files changed, 312 insertions(+), 370 deletions(-)
diff --git a/docs/api-reference/automatic-compaction-api.md b/docs/api-reference/automatic-compaction-api.md
index 3ad90c9d339..c09183c0968 100644
--- a/docs/api-reference/automatic-compaction-api.md
+++ b/docs/api-reference/automatic-compaction-api.md
@@ -258,6 +258,102 @@ Host: http://ROUTER_IP:ROUTER_PORT
A successful request returns an HTTP `200 OK` message code and an empty response body.
+### Update cluster-level compaction config
+
+Updates cluster-level configuration for compaction tasks which applies to all datasources, unless explicitly overridden in the datasource compaction config.
+This includes the following fields:
+
+|Config|Description|Default value|
+|------|-----------|-------------|
+|`compactionTaskSlotRatio`|Ratio of number of slots taken up by compaction tasks to the number of total task slots across all workers.|0.1|
+|`maxCompactionTaskSlots`|Maximum number of task slots that can be taken up by compaction tasks and sub-tasks. Minimum number of task slots available for compaction is 1. When using MSQ engine or Native engine with range partitioning, a single compaction job occupies more than one task slot. In this case, the minimum is 2 so that at least one compaction job can always run in the cluster.|2147483647 (i.e. total task slots)|
+|`compactionPolicy`|Policy to choose intervals for compaction. Currently, the only supported policy is [Newest segment first](#compaction-policy-newestsegmentfirst).|Newest segment first|
+|`useSupervisors`|Whether compaction should be run on Overlord using supervisors instead of Coordinator duties.|false|
+|`engine`|Engine used for running compaction tasks, unless overridden in the datasource-level compaction config. Possible values are `native` and `msq`. `msq` engine can be used for compaction only if `useSupervisors` is `true`.|`native`|
+
+#### Compaction policy `newestSegmentFirst`
+
+|Field|Description|Default value|
+|-----|-----------|-------------|
+|`type`|This must always be `newestSegmentFirst`||
+|`priorityDatasource`|Datasource to prioritize for compaction. The intervals of this datasource are chosen for compaction before the intervals of any other datasource. Within this datasource, the intervals are prioritized based on the chosen compaction policy.|None|
+
+
+#### URL
+
+`POST` `/druid/coordinator/v1/config/compaction/cluster`
+
+#### Responses
+
+<Tabs>
+
+<TabItem value="8" label="200 SUCCESS">
+
+
+*Successfully updated compaction configuration*
+
+</TabItem>
+<TabItem value="9" label="404 NOT FOUND">
+
+
+*Invalid `max` value*
+
+</TabItem>
+</Tabs>
+
+---
+
+#### Sample request
+
+<Tabs>
+
+<TabItem value="10" label="cURL">
+
+
+```shell
+curl --request POST "http://ROUTER_IP:ROUTER_PORT/druid/coordinator/v1/config/compaction/cluster" \
+--header 'Content-Type: application/json' \
+--data '{
+ "compactionTaskSlotRatio": 0.5,
+ "maxCompactionTaskSlots": 1500,
+ "compactionPolicy": {
+ "type": "newestSegmentFirst",
+ "priorityDatasource": "wikipedia"
+ },
+ "useSupervisors": true,
+ "engine": "msq"
+}'
+
+```
+
+</TabItem>
+<TabItem value="11" label="HTTP">
+
+
+```HTTP
+POST /druid/coordinator/v1/config/compaction/cluster HTTP/1.1
+Host: http://ROUTER_IP:ROUTER_PORT
+Content-Type: application/json
+
+{
+ "compactionTaskSlotRatio": 0.5,
+ "maxCompactionTaskSlots": 1500,
+ "compactionPolicy": {
+ "type": "newestSegmentFirst",
+ "priorityDatasource": "wikipedia"
+ },
+ "useSupervisors": true,
+ "engine": "msq"
+}
+```
+
+</TabItem>
+</Tabs>
+
+#### Sample response
+
+A successful request returns an HTTP `200 OK` message code and an empty response body.
+
## View automatic compaction configuration
### Get all automatic compaction configurations
@@ -406,7 +502,7 @@ Host: http://ROUTER_IP:ROUTER_PORT
],
"compactionTaskSlotRatio": 0.1,
"maxCompactionTaskSlots": 2147483647,
- "useAutoScaleSlots": false
+
}
```
</details>
@@ -593,7 +689,12 @@ Host: http://ROUTER_IP:ROUTER_PORT
"globalConfig": {
"compactionTaskSlotRatio": 0.1,
"maxCompactionTaskSlots": 2147483647,
- "useAutoScaleSlots": false
+ "compactionPolicy": {
+ "type": "newestSegmentFirst",
+ "priorityDatasource": "wikipedia"
+ },
+ "useSupervisors": true,
+ "engine": "native"
},
"compactionConfig": {
"dataSource": "wikipedia_hour",
@@ -624,7 +725,11 @@ Host: http://ROUTER_IP:ROUTER_PORT
"globalConfig": {
"compactionTaskSlotRatio": 0.1,
"maxCompactionTaskSlots": 2147483647,
- "useAutoScaleSlots": false
+ "compactionPolicy": {
+ "type": "newestSegmentFirst"
+ },
+ "useSupervisors": false,
+ "engine": "native"
},
"compactionConfig": {
"dataSource": "wikipedia_hour",
diff --git a/docs/data-management/automatic-compaction.md b/docs/data-management/automatic-compaction.md
index 62e209317f8..7d16eae2618 100644
--- a/docs/data-management/automatic-compaction.md
+++ b/docs/data-management/automatic-compaction.md
@@ -243,11 +243,12 @@ You can run automatic compaction using compaction supervisors on the Overlord ra
* Tracked compaction task status to avoid re-compacting an interval repeatedly
-To use compaction supervisors, set the following properties in your Overlord runtime properties:
- * `druid.supervisor.compaction.enabled` to `true` so that compaction tasks can be run as supervisor tasks
- * `druid.supervisor.compaction.engine` to `msq` to specify the MSQ task engine as the compaction engine or to `native` to use the native engine. This is the default engine if the `engine` field is omitted from your compaction config
+To use compaction supervisors, update the [compaction dynamic config](../api-reference/automatic-compaction-api.md#update-cluster-level-compaction-config) and set:
-Compaction supervisors use the same syntax as auto-compaction using Coordinator duties with one key difference: you submit the auto-compaction as a a supervisor spec. In the spec, set the `type` to `autocompact` and include the auto-compaction config in the `spec`.
+* `useSupervisors` to `true` so that compaction tasks can be run as supervisor tasks
+* `engine` to `msq` to use the MSQ task engine as the compaction engine or to `native` (default value) to use the native engine.
+
+Compaction supervisors use the same syntax as auto-compaction using Coordinator duties with one key difference: you submit the auto-compaction as a supervisor spec. In the spec, set the `type` to `autocompact` and include the auto-compaction config in the `spec`.
To submit an automatic compaction task, you can submit a supervisor spec through the [web console](#manage-compaction-supervisors-with-the-web-console) or the [supervisor API](#manage-compaction-supervisors-with-supervisor-apis).
diff --git a/indexing-service/src/main/java/org/apache/druid/guice/SupervisorModule.java b/indexing-service/src/main/java/org/apache/druid/guice/SupervisorModule.java
index 6f510dc562b..73e8e06e896 100644
--- a/indexing-service/src/main/java/org/apache/druid/guice/SupervisorModule.java
+++ b/indexing-service/src/main/java/org/apache/druid/guice/SupervisorModule.java
@@ -29,7 +29,6 @@ import org.apache.druid.indexing.compact.CompactionSupervisorSpec;
import org.apache.druid.indexing.overlord.supervisor.SupervisorStateManagerConfig;
import org.apache.druid.indexing.scheduledbatch.ScheduledBatchSupervisorSpec;
import org.apache.druid.initialization.DruidModule;
-import org.apache.druid.server.coordinator.CompactionSupervisorConfig;
import java.util.List;
@@ -39,7 +38,6 @@ public class SupervisorModule implements DruidModule
public void configure(Binder binder)
{
JsonConfigProvider.bind(binder, "druid.supervisor", SupervisorStateManagerConfig.class);
- JsonConfigProvider.bind(binder, "druid.supervisor.compaction", CompactionSupervisorConfig.class);
}
@Override
diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/compact/CompactionScheduler.java b/indexing-service/src/main/java/org/apache/druid/indexing/compact/CompactionScheduler.java
index 83a87b358c9..0df084f362c 100644
--- a/indexing-service/src/main/java/org/apache/druid/indexing/compact/CompactionScheduler.java
+++ b/indexing-service/src/main/java/org/apache/druid/indexing/compact/CompactionScheduler.java
@@ -23,14 +23,14 @@ import org.apache.druid.server.compaction.CompactionSimulateResult;
import org.apache.druid.server.coordinator.AutoCompactionSnapshot;
import org.apache.druid.server.coordinator.ClusterCompactionConfig;
import org.apache.druid.server.coordinator.CompactionConfigValidationResult;
-import org.apache.druid.server.coordinator.CompactionSupervisorConfig;
import org.apache.druid.server.coordinator.DataSourceCompactionConfig;
+import org.apache.druid.server.coordinator.DruidCompactionConfig;
import java.util.Map;
/**
- * Compaction scheduler that runs on the Overlord if {@link CompactionSupervisorConfig}
- * is enabled.
+ * Compaction scheduler that runs on the Overlord if
+ * {@link DruidCompactionConfig#isUseSupervisors()} is true.
* <p>
* Usage:
* <ul>
@@ -48,6 +48,8 @@ public interface CompactionScheduler
void stop();
+ boolean isEnabled();
+
boolean isRunning();
CompactionConfigValidationResult validateCompactionConfig(DataSourceCompactionConfig compactionConfig);
diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/compact/OverlordCompactionScheduler.java b/indexing-service/src/main/java/org/apache/druid/indexing/compact/OverlordCompactionScheduler.java
index 020c0a0ac22..83a2c5f9f57 100644
--- a/indexing-service/src/main/java/org/apache/druid/indexing/compact/OverlordCompactionScheduler.java
+++ b/indexing-service/src/main/java/org/apache/druid/indexing/compact/OverlordCompactionScheduler.java
@@ -44,7 +44,6 @@ import org.apache.druid.server.compaction.CompactionStatusTracker;
import org.apache.druid.server.coordinator.AutoCompactionSnapshot;
import org.apache.druid.server.coordinator.ClusterCompactionConfig;
import org.apache.druid.server.coordinator.CompactionConfigValidationResult;
-import org.apache.druid.server.coordinator.CompactionSupervisorConfig;
import org.apache.druid.server.coordinator.CoordinatorOverlordServiceConfig;
import org.apache.druid.server.coordinator.DataSourceCompactionConfig;
import org.apache.druid.server.coordinator.DruidCompactionConfig;
@@ -66,7 +65,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
/**
* Implementation of {@link CompactionScheduler}.
* <p>
- * When {@link CompactionSupervisorConfig} is enabled, this class performs the
+ * When {@link DruidCompactionConfig#isUseSupervisors()} is true, this class performs the
* following responsibilities on the leader Overlord:
* <ul>
* <li>Poll segments from metadata</li>
@@ -88,7 +87,6 @@ public class OverlordCompactionScheduler implements CompactionScheduler
private final ServiceEmitter emitter;
private final TaskMaster taskMaster;
- private final CompactionSupervisorConfig supervisorConfig;
private final Supplier<DruidCompactionConfig> compactionConfigSupplier;
private final ConcurrentHashMap<String, DataSourceCompactionConfig> activeDatasourceConfigs;
@@ -124,7 +122,6 @@ public class OverlordCompactionScheduler implements CompactionScheduler
SegmentsMetadataManager segmentManager,
Supplier<DruidCompactionConfig> compactionConfigSupplier,
CompactionStatusTracker statusTracker,
- CompactionSupervisorConfig supervisorConfig,
CoordinatorOverlordServiceConfig coordinatorOverlordServiceConfig,
ScheduledExecutorFactory executorFactory,
ServiceEmitter emitter,
@@ -134,7 +131,6 @@ public class OverlordCompactionScheduler implements CompactionScheduler
this.segmentManager = segmentManager;
this.emitter = emitter;
this.taskMaster = taskMaster;
- this.supervisorConfig = supervisorConfig;
this.compactionConfigSupplier = compactionConfigSupplier;
this.executor = executorFactory.create(1, "CompactionScheduler-%s");
@@ -202,7 +198,7 @@ public class OverlordCompactionScheduler implements CompactionScheduler
} else {
return ClientCompactionRunnerInfo.validateCompactionConfig(
compactionConfig,
- supervisorConfig.getEngine()
+ getLatestConfig().getEngine()
);
}
}
@@ -249,9 +245,10 @@ public class OverlordCompactionScheduler implements CompactionScheduler
}
}
- private boolean isEnabled()
+ @Override
+ public boolean isEnabled()
{
- return supervisorConfig.isEnabled();
+ return compactionConfigSupplier.get().isUseSupervisors();
}
private synchronized void scheduledRun()
@@ -272,7 +269,7 @@ public class OverlordCompactionScheduler implements CompactionScheduler
private synchronized void runCompactionDuty()
{
final CoordinatorRunStats stats = new CoordinatorRunStats();
- duty.run(getLatestConfig(), getDatasourceSnapshot(), supervisorConfig.getEngine(), stats);
+ duty.run(getLatestConfig(), getDatasourceSnapshot(), getLatestConfig().getEngine(), stats);
// Emit stats only if emission period has elapsed
if (!sinceStatsEmitted.isRunning() || sinceStatsEmitted.hasElapsed(METRIC_EMISSION_PERIOD)) {
@@ -310,7 +307,7 @@ public class OverlordCompactionScheduler implements CompactionScheduler
return new CompactionRunSimulator(statusTracker, overlordClient).simulateRunWithConfig(
getLatestConfig().withClusterConfig(updateRequest),
getDatasourceSnapshot(),
- supervisorConfig.getEngine()
+ getLatestConfig().getEngine()
);
} else {
return new CompactionSimulateResult(Collections.emptyMap());
diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/http/OverlordCompactionResource.java b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/http/OverlordCompactionResource.java
index 425dfc9f35f..b29cefef83c 100644
--- a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/http/OverlordCompactionResource.java
+++ b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/http/OverlordCompactionResource.java
@@ -29,7 +29,6 @@ import org.apache.druid.server.compaction.CompactionProgressResponse;
import org.apache.druid.server.compaction.CompactionStatusResponse;
import org.apache.druid.server.coordinator.AutoCompactionSnapshot;
import org.apache.druid.server.coordinator.ClusterCompactionConfig;
-import org.apache.druid.server.coordinator.CompactionSupervisorConfig;
import org.apache.druid.server.http.ServletResourceUtils;
import org.apache.druid.server.http.security.StateResourceFilter;
@@ -52,16 +51,13 @@ import java.util.Collections;
public class OverlordCompactionResource
{
private final CompactionScheduler scheduler;
- private final CompactionSupervisorConfig supervisorConfig;
@Inject
public OverlordCompactionResource(
- CompactionSupervisorConfig supervisorConfig,
CompactionScheduler scheduler
)
{
this.scheduler = scheduler;
- this.supervisorConfig = supervisorConfig;
}
@GET
@@ -70,7 +66,7 @@ public class OverlordCompactionResource
@ResourceFilters(StateResourceFilter.class)
public Response isCompactionSupervisorEnabled()
{
- return Response.ok(supervisorConfig.isEnabled()).build();
+ return Response.ok(scheduler.isEnabled()).build();
}
@GET
@@ -81,7 +77,7 @@ public class OverlordCompactionResource
@QueryParam("dataSource") String dataSource
)
{
- if (!supervisorConfig.isEnabled()) {
+ if (!scheduler.isEnabled()) {
return buildErrorResponseIfSchedulerDisabled();
}
@@ -106,7 +102,7 @@ public class OverlordCompactionResource
@QueryParam("dataSource") String dataSource
)
{
- if (!supervisorConfig.isEnabled()) {
+ if (!scheduler.isEnabled()) {
return buildErrorResponseIfSchedulerDisabled();
}
diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/compact/OverlordCompactionSchedulerTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/compact/OverlordCompactionSchedulerTest.java
index 2805b05cc6f..c627a0a4d78 100644
--- a/indexing-service/src/test/java/org/apache/druid/indexing/compact/OverlordCompactionSchedulerTest.java
+++ b/indexing-service/src/test/java/org/apache/druid/indexing/compact/OverlordCompactionSchedulerTest.java
@@ -51,7 +51,6 @@ import org.apache.druid.server.compaction.Table;
import org.apache.druid.server.coordinator.AutoCompactionSnapshot;
import org.apache.druid.server.coordinator.ClusterCompactionConfig;
import org.apache.druid.server.coordinator.CompactionConfigValidationResult;
-import org.apache.druid.server.coordinator.CompactionSupervisorConfig;
import org.apache.druid.server.coordinator.CoordinatorOverlordServiceConfig;
import org.apache.druid.server.coordinator.CreateDataSegments;
import org.apache.druid.server.coordinator.DataSourceCompactionConfig;
@@ -90,7 +89,6 @@ public class OverlordCompactionSchedulerTest
);
}
- private CompactionSupervisorConfig supervisorConfig;
private DruidCompactionConfig compactionConfig;
private CoordinatorOverlordServiceConfig coordinatorOverlordServiceConfig;
@@ -128,14 +126,13 @@ public class OverlordCompactionSchedulerTest
serviceEmitter = new StubServiceEmitter();
segmentsMetadataManager = new TestSegmentsMetadataManager();
- supervisorConfig = new CompactionSupervisorConfig(true, null);
compactionConfig = DruidCompactionConfig.empty();
coordinatorOverlordServiceConfig = new CoordinatorOverlordServiceConfig(false, null);
- initScheduler();
+ initScheduler(true);
}
- private void initScheduler()
+ private void initScheduler(boolean isSupervisorEnabled)
{
TaskLockbox taskLockbox = new TaskLockbox(taskStorage, new TestIndexerMetadataStorageCoordinator());
WorkerBehaviorConfig defaultWorkerConfig
@@ -144,9 +141,10 @@ public class OverlordCompactionSchedulerTest
taskMaster,
new TaskQueryTool(taskStorage, taskLockbox, taskMaster, null, () -> defaultWorkerConfig),
segmentsMetadataManager,
- () -> compactionConfig,
+ () -> compactionConfig.withClusterConfig(
+ new ClusterCompactionConfig(null, null, null, isSupervisorEnabled, null)
+ ),
new CompactionStatusTracker(OBJECT_MAPPER),
- supervisorConfig,
coordinatorOverlordServiceConfig,
(nameFormat, numThreads) -> new WrappingScheduledExecutorService("test", executor, false),
serviceEmitter,
@@ -157,7 +155,6 @@ public class OverlordCompactionSchedulerTest
@Test
public void testStartStopWhenSchedulerIsEnabled()
{
- supervisorConfig = new CompactionSupervisorConfig(true, null);
Assert.assertFalse(scheduler.isRunning());
scheduler.start();
@@ -176,8 +173,7 @@ public class OverlordCompactionSchedulerTest
@Test
public void testStartStopWhenScheduledIsDisabled()
{
- supervisorConfig = new CompactionSupervisorConfig(false, null);
- initScheduler();
+ initScheduler(false);
Assert.assertFalse(scheduler.isRunning());
scheduler.start();
@@ -191,8 +187,7 @@ public class OverlordCompactionSchedulerTest
@Test
public void testSegmentsAreNotPolledWhenSchedulerIsDisabled()
{
- supervisorConfig = new CompactionSupervisorConfig(false, null);
- initScheduler();
+ initScheduler(false);
verifySegmentPolling(false);
}
@@ -201,7 +196,7 @@ public class OverlordCompactionSchedulerTest
public void testSegmentsArePolledWhenRunningInStandaloneMode()
{
coordinatorOverlordServiceConfig = new CoordinatorOverlordServiceConfig(false, null);
- initScheduler();
+ initScheduler(true);
verifySegmentPolling(true);
}
@@ -210,7 +205,7 @@ public class OverlordCompactionSchedulerTest
public void testSegmentsAreNotPolledWhenRunningInCoordinatorMode()
{
coordinatorOverlordServiceConfig = new CoordinatorOverlordServiceConfig(true, "overlord");
- initScheduler();
+ initScheduler(true);
verifySegmentPolling(false);
}
@@ -345,7 +340,7 @@ public class OverlordCompactionSchedulerTest
);
final CompactionSimulateResult simulateResult = scheduler.simulateRunWithConfigUpdate(
- new ClusterCompactionConfig(null, null, null, null)
+ new ClusterCompactionConfig(null, null, null, null, null)
);
Assert.assertEquals(1, simulateResult.getCompactionStates().size());
final Table pendingCompactionTable = simulateResult.getCompactionStates().get(CompactionStatus.State.PENDING);
@@ -370,7 +365,7 @@ public class OverlordCompactionSchedulerTest
scheduler.stopCompaction(TestDataSource.WIKI);
final CompactionSimulateResult simulateResultWhenDisabled = scheduler.simulateRunWithConfigUpdate(
- new ClusterCompactionConfig(null, null, null, null)
+ new ClusterCompactionConfig(null, null, null, null, null)
);
Assert.assertTrue(simulateResultWhenDisabled.getCompactionStates().isEmpty());
diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/http/OverlordCompactionResourceTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/http/OverlordCompactionResourceTest.java
index d0aba195c42..de9cd7ec6c0 100644
--- a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/http/OverlordCompactionResourceTest.java
+++ b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/http/OverlordCompactionResourceTest.java
@@ -29,7 +29,6 @@ import org.apache.druid.server.compaction.CompactionProgressResponse;
import org.apache.druid.server.compaction.CompactionStatistics;
import org.apache.druid.server.compaction.CompactionStatusResponse;
import org.apache.druid.server.coordinator.AutoCompactionSnapshot;
-import org.apache.druid.server.coordinator.CompactionSupervisorConfig;
import org.easymock.EasyMock;
import org.hamcrest.MatcherAssert;
import org.junit.After;
@@ -43,17 +42,13 @@ import java.util.Map;
public class OverlordCompactionResourceTest
{
- private static final CompactionSupervisorConfig SUPERVISOR_ENABLED
- = new CompactionSupervisorConfig(true, null);
- private static final CompactionSupervisorConfig SUPERVISOR_DISABLED
- = new CompactionSupervisorConfig(false, null);
-
private CompactionScheduler scheduler;
@Before
public void setUp()
{
scheduler = EasyMock.createStrictMock(CompactionScheduler.class);
+ EasyMock.expect(scheduler.isEnabled()).andReturn(true).anyTimes();
}
@After
@@ -74,7 +69,7 @@ public class OverlordCompactionResourceTest
.andReturn(allSnapshots).once();
EasyMock.replay(scheduler);
- final Response response = new OverlordCompactionResource(SUPERVISOR_ENABLED, scheduler)
+ final Response response = new OverlordCompactionResource(scheduler)
.getCompactionSnapshots("");
Assert.assertEquals(200, response.getStatus());
Assert.assertEquals(
@@ -95,7 +90,7 @@ public class OverlordCompactionResourceTest
.andReturn(allSnapshots).once();
EasyMock.replay(scheduler);
- final Response response = new OverlordCompactionResource(SUPERVISOR_ENABLED, scheduler)
+ final Response response = new OverlordCompactionResource(scheduler)
.getCompactionSnapshots(null);
Assert.assertEquals(200, response.getStatus());
Assert.assertEquals(
@@ -113,7 +108,7 @@ public class OverlordCompactionResourceTest
.andReturn(snapshot).once();
EasyMock.replay(scheduler);
- final Response response = new OverlordCompactionResource(SUPERVISOR_ENABLED, scheduler)
+ final Response response = new OverlordCompactionResource(scheduler)
.getCompactionSnapshots(TestDataSource.WIKI);
Assert.assertEquals(200, response.getStatus());
Assert.assertEquals(
@@ -129,7 +124,7 @@ public class OverlordCompactionResourceTest
.andReturn(null).once();
EasyMock.replay(scheduler);
- final Response response = new OverlordCompactionResource(SUPERVISOR_ENABLED, scheduler)
+ final Response response = new OverlordCompactionResource(scheduler)
.getCompactionSnapshots(TestDataSource.KOALA);
Assert.assertEquals(404, response.getStatus());
}
@@ -146,7 +141,7 @@ public class OverlordCompactionResourceTest
.andReturn(snapshot).once();
EasyMock.replay(scheduler);
- final Response response = new OverlordCompactionResource(SUPERVISOR_ENABLED, scheduler)
+ final Response response = new OverlordCompactionResource(scheduler)
.getCompactionProgress(TestDataSource.WIKI);
Assert.assertEquals(200, response.getStatus());
Assert.assertEquals(new CompactionProgressResponse(100L), response.getEntity());
@@ -157,7 +152,7 @@ public class OverlordCompactionResourceTest
{
EasyMock.replay(scheduler);
- final Response response = new OverlordCompactionResource(SUPERVISOR_ENABLED, scheduler)
+ final Response response = new OverlordCompactionResource(scheduler)
.getCompactionProgress(null);
Assert.assertEquals(Response.Status.BAD_REQUEST.getStatusCode(), response.getStatus());
@@ -177,7 +172,7 @@ public class OverlordCompactionResourceTest
.andReturn(null).once();
EasyMock.replay(scheduler);
- final Response response = new OverlordCompactionResource(SUPERVISOR_ENABLED, scheduler)
+ final Response response = new OverlordCompactionResource(scheduler)
.getCompactionProgress(TestDataSource.KOALA);
Assert.assertEquals(Response.Status.NOT_FOUND.getStatusCode(), response.getStatus());
@@ -193,9 +188,12 @@ public class OverlordCompactionResourceTest
@Test
public void testGetProgressReturnsUnsupportedWhenSupervisorDisabled()
{
+ scheduler = EasyMock.createStrictMock(CompactionScheduler.class);
+ EasyMock.expect(scheduler.isEnabled()).andReturn(false).once();
EasyMock.replay(scheduler);
+
verifyResponseWhenSupervisorDisabled(
- new OverlordCompactionResource(SUPERVISOR_DISABLED, scheduler)
+ new OverlordCompactionResource(scheduler)
.getCompactionProgress(TestDataSource.WIKI)
);
}
@@ -203,9 +201,12 @@ public class OverlordCompactionResourceTest
@Test
public void testGetSnapshotReturnsUnsupportedWhenSupervisorDisabled()
{
+ scheduler = EasyMock.createStrictMock(CompactionScheduler.class);
+ EasyMock.expect(scheduler.isEnabled()).andReturn(false).once();
EasyMock.replay(scheduler);
+
verifyResponseWhenSupervisorDisabled(
- new OverlordCompactionResource(SUPERVISOR_DISABLED, scheduler)
+ new OverlordCompactionResource(scheduler)
.getCompactionSnapshots(TestDataSource.WIKI)
);
}
diff --git a/integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/utils/DruidClusterAdminClient.java b/integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/utils/DruidClusterAdminClient.java
index 9c15eef30c5..48b7e4b6236 100644
--- a/integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/utils/DruidClusterAdminClient.java
+++ b/integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/utils/DruidClusterAdminClient.java
@@ -215,7 +215,7 @@ public class DruidClusterAdminClient
.exec(new ExecStartResultCallback(stdout, stderr))
.awaitCompletion();
- return new Pair<>(stdout.toString(StandardCharsets.UTF_8.name()), stderr.toString(StandardCharsets.UTF_8.name()));
+ return new Pair<>(stdout.toString(StandardCharsets.UTF_8), stderr.toString(StandardCharsets.UTF_8));
}
public void restartDockerContainer(String serviceName)
diff --git a/integration-tests/src/main/java/org/apache/druid/testing/clients/CompactionResourceTestClient.java b/integration-tests/src/main/java/org/apache/druid/testing/clients/CompactionResourceTestClient.java
index 849af497567..da0b2f3516d 100644
--- a/integration-tests/src/main/java/org/apache/druid/testing/clients/CompactionResourceTestClient.java
+++ b/integration-tests/src/main/java/org/apache/druid/testing/clients/CompactionResourceTestClient.java
@@ -145,21 +145,14 @@ public class CompactionResourceTestClient
}
}
- public void updateCompactionTaskSlot(Double compactionTaskSlotRatio, Integer maxCompactionTaskSlots, Boolean useAutoScaleSlots) throws Exception
+ public void updateCompactionTaskSlot(Double compactionTaskSlotRatio, Integer maxCompactionTaskSlots) throws Exception
{
- String url;
- if (useAutoScaleSlots == null) {
- url = StringUtils.format("%sconfig/compaction/taskslots?ratio=%s&max=%s",
- getCoordinatorURL(),
- StringUtils.urlEncode(compactionTaskSlotRatio.toString()),
- StringUtils.urlEncode(maxCompactionTaskSlots.toString()));
- } else {
- url = StringUtils.format("%sconfig/compaction/taskslots?ratio=%s&max=%s&useAutoScaleSlots=%s",
- getCoordinatorURL(),
- StringUtils.urlEncode(compactionTaskSlotRatio.toString()),
- StringUtils.urlEncode(maxCompactionTaskSlots.toString()),
- StringUtils.urlEncode(useAutoScaleSlots.toString()));
- }
+ String url = StringUtils.format(
+ "%sconfig/compaction/taskslots?ratio=%s&max=%s",
+ getCoordinatorURL(),
+ StringUtils.urlEncode(compactionTaskSlotRatio.toString()),
+ StringUtils.urlEncode(maxCompactionTaskSlots.toString())
+ );
StatusResponseHolder response = httpClient.go(new Request(HttpMethod.POST, new URL(url)), responseHandler).get();
if (!response.getStatus().equals(HttpResponseStatus.OK)) {
throw new ISE(
diff --git a/integration-tests/src/main/java/org/apache/druid/testing/utils/DruidClusterAdminClient.java b/integration-tests/src/main/java/org/apache/druid/testing/utils/DruidClusterAdminClient.java
index 6b98306ca0e..bc65fb3d8db 100644
--- a/integration-tests/src/main/java/org/apache/druid/testing/utils/DruidClusterAdminClient.java
+++ b/integration-tests/src/main/java/org/apache/druid/testing/utils/DruidClusterAdminClient.java
@@ -214,7 +214,7 @@ public class DruidClusterAdminClient
.exec(new ExecStartResultCallback(stdout, stderr))
.awaitCompletion();
- return new Pair<>(stdout.toString(StandardCharsets.UTF_8.name()), stderr.toString(StandardCharsets.UTF_8.name()));
+ return new Pair<>(stdout.toString(StandardCharsets.UTF_8), stderr.toString(StandardCharsets.UTF_8));
}
public void restartDockerContainer(String serviceName)
diff --git a/integration-tests/src/test/java/org/apache/druid/tests/coordinator/duty/ITAutoCompactionLockContentionTest.java b/integration-tests/src/test/java/org/apache/druid/tests/coordinator/duty/ITAutoCompactionLockContentionTest.java
index 0a1071c38e7..a5943aacb33 100644
--- a/integration-tests/src/test/java/org/apache/druid/tests/coordinator/duty/ITAutoCompactionLockContentionTest.java
+++ b/integration-tests/src/test/java/org/apache/druid/tests/coordinator/duty/ITAutoCompactionLockContentionTest.java
@@ -310,7 +310,7 @@ public class ITAutoCompactionLockContentionTest extends AbstractKafkaIndexingSer
.withSkipOffsetFromLatest(Period.ZERO)
.withMaxRowsPerSegment(Specs.MAX_ROWS_PER_SEGMENT)
.build();
- compactionResource.updateCompactionTaskSlot(0.5, 10, null);
+ compactionResource.updateCompactionTaskSlot(0.5, 10);
compactionResource.submitCompactionConfig(dataSourceCompactionConfig);
// Wait for compaction config to persist
diff --git a/integration-tests/src/test/java/org/apache/druid/tests/coordinator/duty/ITAutoCompactionTest.java b/integration-tests/src/test/java/org/apache/druid/tests/coordinator/duty/ITAutoCompactionTest.java
index baf7177b373..e993dafdbcf 100644
--- a/integration-tests/src/test/java/org/apache/druid/tests/coordinator/duty/ITAutoCompactionTest.java
+++ b/integration-tests/src/test/java/org/apache/druid/tests/coordinator/duty/ITAutoCompactionTest.java
@@ -133,7 +133,7 @@ public class ITAutoCompactionTest extends AbstractIndexerTest
public void setup() throws Exception
{
// Set compaction slot to 5
- updateCompactionTaskSlot(0.5, 10, null);
+ updateCompactionTaskSlot(0.5, 10);
fullDatasourceName = "wikipedia_index_test_" + UUID.randomUUID() + config.getExtraDatasourceNameSuffix();
}
@@ -410,7 +410,7 @@ public class ITAutoCompactionTest extends AbstractIndexerTest
// added = 31
loadData(INDEX_TASK_WITHOUT_ROLLUP_FOR_PRESERVE_METRICS);
if (engine == CompactionEngine.MSQ) {
- updateCompactionTaskSlot(0.1, 2, false);
+ updateCompactionTaskSlot(0.1, 2);
}
try (final Closeable ignored = unloader(fullDatasourceName)) {
final List<String> intervalsBeforeCompaction = coordinator.getSegmentIntervals(fullDatasourceName);
@@ -744,7 +744,7 @@ public class ITAutoCompactionTest extends AbstractIndexerTest
public void testAutoCompactionDutyCanUpdateTaskSlots() throws Exception
{
// Set compactionTaskSlotRatio to 0 to prevent any compaction
- updateCompactionTaskSlot(0, 0, null);
+ updateCompactionTaskSlot(0, 0);
loadData(INDEX_TASK);
try (final Closeable ignored = unloader(fullDatasourceName)) {
final List<String> intervalsBeforeCompaction = coordinator.getSegmentIntervals(fullDatasourceName);
@@ -761,7 +761,7 @@ public class ITAutoCompactionTest extends AbstractIndexerTest
checkCompactionIntervals(intervalsBeforeCompaction);
Assert.assertNull(compactionResource.getCompactionStatus(fullDatasourceName));
// Update compaction slots to be 1
- updateCompactionTaskSlot(1, 1, null);
+ updateCompactionTaskSlot(1, 1);
// One day compacted (1 new segment) and one day remains uncompacted. (3 total)
forceTriggerAutoCompaction(3);
verifyQuery(INDEX_QUERIES_RESOURCE);
@@ -1365,7 +1365,7 @@ public class ITAutoCompactionTest extends AbstractIndexerTest
@Test
public void testAutoCompactionDutyWithSegmentGranularityFinerAndNotAlignWithSegment() throws Exception
{
- updateCompactionTaskSlot(1, 1, null);
+ updateCompactionTaskSlot(1, 1);
final ISOChronology chrono = ISOChronology.getInstance(DateTimes.inferTzFromString("America/Los_Angeles"));
Map<String, Object> specs = ImmutableMap.of("%%GRANULARITYSPEC%%", new UniformGranularitySpec(Granularities.MONTH, Granularities.DAY, false, ImmutableList.of(new Interval("2013-08-31/2013-09-02", chrono))));
loadData(INDEX_TASK_WITH_GRANULARITY_SPEC, specs);
@@ -1416,7 +1416,7 @@ public class ITAutoCompactionTest extends AbstractIndexerTest
@Test(dataProvider = "engine")
public void testAutoCompactionDutyWithSegmentGranularityCoarserAndNotAlignWithSegment(CompactionEngine engine) throws Exception
{
- updateCompactionTaskSlot(1, 1, null);
+ updateCompactionTaskSlot(1, 1);
final ISOChronology chrono = ISOChronology.getInstance(DateTimes.inferTzFromString("America/Los_Angeles"));
Map<String, Object> specs = ImmutableMap.of("%%GRANULARITYSPEC%%", new UniformGranularitySpec(Granularities.WEEK, Granularities.DAY, false, ImmutableList.of(new Interval("2013-08-31/2013-09-02", chrono))));
loadData(INDEX_TASK_WITH_GRANULARITY_SPEC, specs);
@@ -1745,24 +1745,6 @@ public class ITAutoCompactionTest extends AbstractIndexerTest
}
}
- @Test
- public void testUpdateCompactionTaskSlotWithUseAutoScaleSlots() throws Exception
- {
- // First try update without useAutoScaleSlots
- updateCompactionTaskSlot(3, 5, null);
- DruidCompactionConfig compactionConfig = compactionResource.getCompactionConfig();
- // Should be default value which is false
- Assert.assertFalse(compactionConfig.isUseAutoScaleSlots());
- // Now try update from default value to useAutoScaleSlots=true
- updateCompactionTaskSlot(3, 5, true);
- compactionConfig = compactionResource.getCompactionConfig();
- Assert.assertTrue(compactionConfig.isUseAutoScaleSlots());
- // Now try update from useAutoScaleSlots=true to useAutoScaleSlots=false
- updateCompactionTaskSlot(3, 5, false);
- compactionConfig = compactionResource.getCompactionConfig();
- Assert.assertFalse(compactionConfig.isUseAutoScaleSlots());
- }
-
private void loadData(String indexTask) throws Exception
{
loadData(indexTask, ImmutableMap.of());
@@ -2062,16 +2044,13 @@ public class ITAutoCompactionTest extends AbstractIndexerTest
}
}
- private void updateCompactionTaskSlot(double compactionTaskSlotRatio, int maxCompactionTaskSlots, Boolean useAutoScaleSlots) throws Exception
+ private void updateCompactionTaskSlot(double compactionTaskSlotRatio, int maxCompactionTaskSlots) throws Exception
{
- compactionResource.updateCompactionTaskSlot(compactionTaskSlotRatio, maxCompactionTaskSlots, useAutoScaleSlots);
+ compactionResource.updateCompactionTaskSlot(compactionTaskSlotRatio, maxCompactionTaskSlots);
// Verify that the compaction config is updated correctly.
DruidCompactionConfig compactionConfig = compactionResource.getCompactionConfig();
Assert.assertEquals(compactionConfig.getCompactionTaskSlotRatio(), compactionTaskSlotRatio);
Assert.assertEquals(compactionConfig.getMaxCompactionTaskSlots(), maxCompactionTaskSlots);
- if (useAutoScaleSlots != null) {
- Assert.assertEquals(compactionConfig.isUseAutoScaleSlots(), useAutoScaleSlots.booleanValue());
- }
}
private void getAndAssertCompactionStatus(
diff --git a/server/src/main/java/org/apache/druid/server/compaction/CompactionRunSimulator.java b/server/src/main/java/org/apache/druid/server/compaction/CompactionRunSimulator.java
index b74735e6aa3..646319eec3b 100644
--- a/server/src/main/java/org/apache/druid/server/compaction/CompactionRunSimulator.java
+++ b/server/src/main/java/org/apache/druid/server/compaction/CompactionRunSimulator.java
@@ -135,7 +135,7 @@ public class CompactionRunSimulator
// 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)
+ new ClusterCompactionConfig(1.0, Integer.MAX_VALUE, null, null, null)
);
final CoordinatorRunStats stats = new CoordinatorRunStats();
diff --git a/server/src/main/java/org/apache/druid/server/coordinator/ClusterCompactionConfig.java b/server/src/main/java/org/apache/druid/server/coordinator/ClusterCompactionConfig.java
index 2e0c070b1ed..7334251da76 100644
--- a/server/src/main/java/org/apache/druid/server/coordinator/ClusterCompactionConfig.java
+++ b/server/src/main/java/org/apache/druid/server/coordinator/ClusterCompactionConfig.java
@@ -21,7 +21,11 @@ package org.apache.druid.server.coordinator;
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.indexer.CompactionEngine;
import org.apache.druid.server.compaction.CompactionCandidateSearchPolicy;
+import org.apache.druid.server.compaction.NewestSegmentFirstPolicy;
import javax.annotation.Nullable;
import java.util.Objects;
@@ -33,51 +37,63 @@ import java.util.Objects;
*/
public class ClusterCompactionConfig
{
+ private static final CompactionCandidateSearchPolicy DEFAULT_COMPACTION_POLICY
+ = new NewestSegmentFirstPolicy(null);
+
private final Double compactionTaskSlotRatio;
private final Integer maxCompactionTaskSlots;
- private final Boolean useAutoScaleSlots;
+ private final Boolean useSupervisors;
+ private final CompactionEngine engine;
private final CompactionCandidateSearchPolicy compactionPolicy;
@JsonCreator
public ClusterCompactionConfig(
@JsonProperty("compactionTaskSlotRatio") @Nullable Double compactionTaskSlotRatio,
@JsonProperty("maxCompactionTaskSlots") @Nullable Integer maxCompactionTaskSlots,
- @JsonProperty("useAutoScaleSlots") @Nullable Boolean useAutoScaleSlots,
- @JsonProperty("compactionPolicy") @Nullable CompactionCandidateSearchPolicy compactionPolicy
+ @JsonProperty("compactionPolicy") @Nullable CompactionCandidateSearchPolicy compactionPolicy,
+ @JsonProperty("useSupervisors") @Nullable Boolean useSupervisors,
+ @JsonProperty("engine") @Nullable CompactionEngine engine
)
{
- this.compactionTaskSlotRatio = compactionTaskSlotRatio;
- this.maxCompactionTaskSlots = maxCompactionTaskSlots;
- this.useAutoScaleSlots = useAutoScaleSlots;
- this.compactionPolicy = compactionPolicy;
+ this.compactionTaskSlotRatio = Configs.valueOrDefault(compactionTaskSlotRatio, 0.1);
+ this.maxCompactionTaskSlots = Configs.valueOrDefault(maxCompactionTaskSlots, Integer.MAX_VALUE);
+ this.compactionPolicy = Configs.valueOrDefault(compactionPolicy, DEFAULT_COMPACTION_POLICY);
+ this.engine = Configs.valueOrDefault(engine, CompactionEngine.NATIVE);
+ this.useSupervisors = Configs.valueOrDefault(useSupervisors, false);
+
+ if (!this.useSupervisors && this.engine == CompactionEngine.MSQ) {
+ throw InvalidInput.exception("MSQ Compaction engine can be used only with compaction supervisors.");
+ }
}
- @Nullable
@JsonProperty
- public Double getCompactionTaskSlotRatio()
+ public double getCompactionTaskSlotRatio()
{
return compactionTaskSlotRatio;
}
- @Nullable
@JsonProperty
- public Integer getMaxCompactionTaskSlots()
+ public int getMaxCompactionTaskSlots()
{
return maxCompactionTaskSlots;
}
- @Nullable
@JsonProperty
- public Boolean getUseAutoScaleSlots()
+ public CompactionCandidateSearchPolicy getCompactionPolicy()
{
- return useAutoScaleSlots;
+ return compactionPolicy;
}
- @Nullable
@JsonProperty
- public CompactionCandidateSearchPolicy getCompactionPolicy()
+ public boolean getUseSupervisors()
{
- return compactionPolicy;
+ return useSupervisors;
+ }
+
+ @JsonProperty
+ public CompactionEngine getEngine()
+ {
+ return engine;
}
@Override
@@ -92,8 +108,9 @@ public class ClusterCompactionConfig
ClusterCompactionConfig that = (ClusterCompactionConfig) o;
return Objects.equals(compactionTaskSlotRatio, that.compactionTaskSlotRatio)
&& Objects.equals(maxCompactionTaskSlots, that.maxCompactionTaskSlots)
- && Objects.equals(useAutoScaleSlots, that.useAutoScaleSlots)
- && Objects.equals(compactionPolicy, that.compactionPolicy);
+ && Objects.equals(compactionPolicy, that.compactionPolicy)
+ && Objects.equals(useSupervisors, that.useSupervisors)
+ && Objects.equals(engine, that.engine);
}
@Override
@@ -102,8 +119,21 @@ public class ClusterCompactionConfig
return Objects.hash(
compactionTaskSlotRatio,
maxCompactionTaskSlots,
- useAutoScaleSlots,
- compactionPolicy
+ compactionPolicy,
+ useSupervisors,
+ engine
);
}
+
+ @Override
+ public String toString()
+ {
+ return "ClusterCompactionConfig{" +
+ "compactionTaskSlotRatio=" + compactionTaskSlotRatio +
+ ", maxCompactionTaskSlots=" + maxCompactionTaskSlots +
+ ", useSupervisors=" + useSupervisors +
+ ", engine=" + engine +
+ ", compactionPolicy=" + compactionPolicy +
+ '}';
+ }
}
diff --git a/server/src/main/java/org/apache/druid/server/coordinator/CompactionSupervisorConfig.java b/server/src/main/java/org/apache/druid/server/coordinator/CompactionSupervisorConfig.java
deleted file mode 100644
index 2cc2f0d133f..00000000000
--- a/server/src/main/java/org/apache/druid/server/coordinator/CompactionSupervisorConfig.java
+++ /dev/null
@@ -1,96 +0,0 @@
-/*
- * 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.coordinator;
-
-import com.fasterxml.jackson.annotation.JsonCreator;
-import com.fasterxml.jackson.annotation.JsonProperty;
-import org.apache.druid.common.config.Configs;
-import org.apache.druid.indexer.CompactionEngine;
-
-import javax.annotation.Nullable;
-import java.util.Objects;
-
-/**
- * This config must be bound on CliOverlord to enable running compaction supervisors
- * on the Overlord. When compaction supervisors are enabled, the Coordinator
- * does not run auto-compact duty.
- */
-public class CompactionSupervisorConfig
-{
- private static final CompactionSupervisorConfig DEFAULT = new CompactionSupervisorConfig(null, null);
-
- @JsonProperty
- private final boolean enabled;
- @JsonProperty
- private final CompactionEngine engine;
-
- public static CompactionSupervisorConfig defaultConfig()
- {
- return DEFAULT;
- }
-
- @JsonCreator
- public CompactionSupervisorConfig(
- @JsonProperty("enabled") @Nullable Boolean enabled,
- @JsonProperty("engine") @Nullable CompactionEngine engine
- )
- {
- this.enabled = Configs.valueOrDefault(enabled, false);
- this.engine = Configs.valueOrDefault(engine, CompactionEngine.NATIVE);
- }
-
- public boolean isEnabled()
- {
- return enabled;
- }
-
- public CompactionEngine getEngine()
- {
- return engine;
- }
-
- @Override
- public boolean equals(Object o)
- {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
- CompactionSupervisorConfig that = (CompactionSupervisorConfig) o;
- return enabled == that.enabled && engine == that.engine;
- }
-
- @Override
- public int hashCode()
- {
- return Objects.hash(enabled, engine);
- }
-
- @Override
- public String toString()
- {
- return "CompactionSchedulerConfig{" +
- "enabled=" + enabled +
- "engine=" + engine +
- '}';
- }
-}
diff --git a/server/src/main/java/org/apache/druid/server/coordinator/DruidCompactionConfig.java b/server/src/main/java/org/apache/druid/server/coordinator/DruidCompactionConfig.java
index 96338f5b2ef..d9298c6ab63 100644
--- a/server/src/main/java/org/apache/druid/server/coordinator/DruidCompactionConfig.java
+++ b/server/src/main/java/org/apache/druid/server/coordinator/DruidCompactionConfig.java
@@ -23,8 +23,8 @@ import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.google.common.base.Optional;
import org.apache.druid.common.config.Configs;
+import org.apache.druid.indexer.CompactionEngine;
import org.apache.druid.server.compaction.CompactionCandidateSearchPolicy;
-import org.apache.druid.server.compaction.NewestSegmentFirstPolicy;
import javax.annotation.Nullable;
import java.util.ArrayList;
@@ -39,41 +39,32 @@ public class DruidCompactionConfig
{
public static final String CONFIG_KEY = "coordinator.compaction.config";
- private static final CompactionCandidateSearchPolicy DEFAULT_COMPACTION_POLICY
- = new NewestSegmentFirstPolicy(null);
private static final DruidCompactionConfig EMPTY_INSTANCE
- = new DruidCompactionConfig(Collections.emptyList(), null, null, null, null);
+ = new DruidCompactionConfig(List.of(), null, null, null, null, null);
private final List<DataSourceCompactionConfig> compactionConfigs;
- private final double compactionTaskSlotRatio;
- private final int maxCompactionTaskSlots;
- private final boolean useAutoScaleSlots;
- private final CompactionCandidateSearchPolicy compactionPolicy;
+ private final ClusterCompactionConfig clusterConfig;
+ /**
+ * Creates a copy of this {@link DruidCompactionConfig} with the given
+ * datasource compaction configs.
+ */
public DruidCompactionConfig withDatasourceConfigs(
List<DataSourceCompactionConfig> compactionConfigs
)
{
- return new DruidCompactionConfig(
- compactionConfigs,
- compactionTaskSlotRatio,
- maxCompactionTaskSlots,
- useAutoScaleSlots,
- compactionPolicy
- );
+ return new DruidCompactionConfig(compactionConfigs, clusterConfig);
}
+ /**
+ * Creates a copy of this {@link DruidCompactionConfig} by updating the
+ * fields provided in the {@link ClusterCompactionConfig}.
+ */
public DruidCompactionConfig withClusterConfig(
- ClusterCompactionConfig update
+ ClusterCompactionConfig clusterConfig
)
{
- return new DruidCompactionConfig(
- this.compactionConfigs,
- Configs.valueOrDefault(update.getCompactionTaskSlotRatio(), compactionTaskSlotRatio),
- Configs.valueOrDefault(update.getMaxCompactionTaskSlots(), maxCompactionTaskSlots),
- Configs.valueOrDefault(update.getUseAutoScaleSlots(), useAutoScaleSlots),
- Configs.valueOrDefault(update.getCompactionPolicy(), compactionPolicy)
- );
+ return new DruidCompactionConfig(compactionConfigs, clusterConfig);
}
public DruidCompactionConfig withDatasourceConfig(DataSourceCompactionConfig dataSourceConfig)
@@ -93,15 +84,30 @@ public class DruidCompactionConfig
@JsonProperty("compactionConfigs") List<DataSourceCompactionConfig> compactionConfigs,
@JsonProperty("compactionTaskSlotRatio") @Nullable Double compactionTaskSlotRatio,
@JsonProperty("maxCompactionTaskSlots") @Nullable Integer maxCompactionTaskSlots,
- @JsonProperty("useAutoScaleSlots") @Nullable Boolean useAutoScaleSlots,
- @JsonProperty("compactionPolicy") @Nullable CompactionCandidateSearchPolicy compactionPolicy
+ @JsonProperty("compactionPolicy") @Nullable CompactionCandidateSearchPolicy compactionPolicy,
+ @JsonProperty("useSupervisors") @Nullable Boolean useSupervisors,
+ @JsonProperty("engine") @Nullable CompactionEngine engine
+ )
+ {
+ this(
+ compactionConfigs,
+ new ClusterCompactionConfig(
+ compactionTaskSlotRatio,
+ maxCompactionTaskSlots,
+ compactionPolicy,
+ useSupervisors,
+ engine
+ )
+ );
+ }
+
+ private DruidCompactionConfig(
+ @Nullable List<DataSourceCompactionConfig> compactionConfigs,
+ @Nullable ClusterCompactionConfig clusterConfig
)
{
this.compactionConfigs = Configs.valueOrDefault(compactionConfigs, Collections.emptyList());
- this.compactionTaskSlotRatio = Configs.valueOrDefault(compactionTaskSlotRatio, 0.1);
- this.maxCompactionTaskSlots = Configs.valueOrDefault(maxCompactionTaskSlots, Integer.MAX_VALUE);
- this.useAutoScaleSlots = Configs.valueOrDefault(useAutoScaleSlots, false);
- this.compactionPolicy = Configs.valueOrDefault(compactionPolicy, DEFAULT_COMPACTION_POLICY);
+ this.clusterConfig = clusterConfig;
}
@JsonProperty
@@ -113,31 +119,33 @@ public class DruidCompactionConfig
@JsonProperty
public double getCompactionTaskSlotRatio()
{
- return compactionTaskSlotRatio;
+ return clusterConfig.getCompactionTaskSlotRatio();
}
@JsonProperty
public int getMaxCompactionTaskSlots()
{
- return maxCompactionTaskSlots;
+ return clusterConfig.getMaxCompactionTaskSlots();
}
@JsonProperty
- public boolean isUseAutoScaleSlots()
+ public boolean isUseSupervisors()
{
- return useAutoScaleSlots;
+ return clusterConfig.getUseSupervisors();
}
+ @JsonProperty
+ public CompactionEngine getEngine()
+ {
+ return clusterConfig.getEngine();
+ }
- // Null-safe getters not used for serialization
+ /**
+ * Returns the cluster-level compaction config. Not used for serialization.
+ */
public ClusterCompactionConfig clusterConfig()
{
- return new ClusterCompactionConfig(
- compactionTaskSlotRatio,
- maxCompactionTaskSlots,
- useAutoScaleSlots,
- compactionPolicy
- );
+ return clusterConfig;
}
public Map<String, DataSourceCompactionConfig> dataSourceToCompactionConfigMap()
@@ -160,7 +168,7 @@ public class DruidCompactionConfig
@JsonProperty
public CompactionCandidateSearchPolicy getCompactionPolicy()
{
- return compactionPolicy;
+ return clusterConfig.getCompactionPolicy();
}
@Override
@@ -173,23 +181,14 @@ public class DruidCompactionConfig
return false;
}
DruidCompactionConfig that = (DruidCompactionConfig) o;
- return Double.compare(that.compactionTaskSlotRatio, compactionTaskSlotRatio) == 0 &&
- maxCompactionTaskSlots == that.maxCompactionTaskSlots &&
- useAutoScaleSlots == that.useAutoScaleSlots &&
- Objects.equals(compactionPolicy, that.compactionPolicy) &&
+ return Objects.equals(clusterConfig, that.clusterConfig) &&
Objects.equals(compactionConfigs, that.compactionConfigs);
}
@Override
public int hashCode()
{
- return Objects.hash(
- compactionConfigs,
- compactionTaskSlotRatio,
- maxCompactionTaskSlots,
- useAutoScaleSlots,
- compactionPolicy
- );
+ return Objects.hash(clusterConfig, compactionConfigs);
}
@Override
@@ -197,10 +196,7 @@ public class DruidCompactionConfig
{
return "CoordinatorCompactionConfig{" +
"compactionConfigs=" + compactionConfigs +
- ", compactionTaskSlotRatio=" + compactionTaskSlotRatio +
- ", maxCompactionTaskSlots=" + maxCompactionTaskSlots +
- ", useAutoScaleSlots=" + useAutoScaleSlots +
- ", compactionPolicy=" + compactionPolicy +
+ ", clusterConfig=" + clusterConfig +
'}';
}
}
diff --git a/server/src/main/java/org/apache/druid/server/http/CoordinatorCompactionConfigsResource.java b/server/src/main/java/org/apache/druid/server/http/CoordinatorCompactionConfigsResource.java
index 5215f7de4e2..f2cad3b455b 100644
--- a/server/src/main/java/org/apache/druid/server/http/CoordinatorCompactionConfigsResource.java
+++ b/server/src/main/java/org/apache/druid/server/http/CoordinatorCompactionConfigsResource.java
@@ -112,7 +112,6 @@ public class CoordinatorCompactionConfigsResource
public Response setCompactionTaskLimit(
@QueryParam("ratio") Double compactionTaskSlotRatio,
@QueryParam("max") Integer maxCompactionTaskSlots,
- @QueryParam("useAutoScaleSlots") Boolean useAutoScaleSlots,
@Context HttpServletRequest req
)
{
@@ -120,7 +119,8 @@ public class CoordinatorCompactionConfigsResource
new ClusterCompactionConfig(
compactionTaskSlotRatio,
maxCompactionTaskSlots,
- useAutoScaleSlots,
+ null,
+ null,
null
),
req
diff --git a/server/src/test/java/org/apache/druid/server/coordinator/CompactionSupervisorConfigTest.java b/server/src/test/java/org/apache/druid/server/coordinator/CompactionSupervisorConfigTest.java
deleted file mode 100644
index 59cc3ecf171..00000000000
--- a/server/src/test/java/org/apache/druid/server/coordinator/CompactionSupervisorConfigTest.java
+++ /dev/null
@@ -1,60 +0,0 @@
-/*
- * 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.coordinator;
-
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.google.common.collect.ImmutableMap;
-import org.apache.druid.indexer.CompactionEngine;
-import org.apache.druid.jackson.DefaultObjectMapper;
-import org.junit.Assert;
-import org.junit.Test;
-
-public class CompactionSupervisorConfigTest
-{
- private static final ObjectMapper OBJECT_MAPPER = new DefaultObjectMapper();
-
- @Test
- public void testCompactionSupervisorConfigSerde() throws JsonProcessingException
- {
- final boolean enabled = true;
- final CompactionEngine defaultEngine = CompactionEngine.MSQ;
- CompactionSupervisorConfig compactionSupervisorConfig =
- OBJECT_MAPPER.readValue(
- OBJECT_MAPPER.writeValueAsString(ImmutableMap.of("enabled", enabled, "engine", defaultEngine)),
- CompactionSupervisorConfig.class
- );
- Assert.assertEquals(new CompactionSupervisorConfig(enabled, defaultEngine), compactionSupervisorConfig);
- }
-
- @Test
- public void testCompactionSupervisorConfigEquality()
- {
- Assert.assertEquals(
- new CompactionSupervisorConfig(true, CompactionEngine.MSQ),
- new CompactionSupervisorConfig(true, CompactionEngine.MSQ)
- );
- Assert.assertNotEquals(
- new CompactionSupervisorConfig(true, CompactionEngine.NATIVE),
- new CompactionSupervisorConfig(true, CompactionEngine.MSQ)
- );
- Assert.assertNotEquals(new CompactionSupervisorConfig(true, CompactionEngine.NATIVE), "true");
- }
-}
diff --git a/server/src/test/java/org/apache/druid/server/coordinator/DataSourceCompactionConfigAuditEntryTest.java b/server/src/test/java/org/apache/druid/server/coordinator/DataSourceCompactionConfigAuditEntryTest.java
index 4647ad8d985..71fd30dbfd2 100644
--- a/server/src/test/java/org/apache/druid/server/coordinator/DataSourceCompactionConfigAuditEntryTest.java
+++ b/server/src/test/java/org/apache/druid/server/coordinator/DataSourceCompactionConfigAuditEntryTest.java
@@ -33,7 +33,7 @@ public class DataSourceCompactionConfigAuditEntryTest
private final AuditInfo auditInfo = new AuditInfo("author", "identity", "comment", "ip");
private final DataSourceCompactionConfigAuditEntry firstEntry = new DataSourceCompactionConfigAuditEntry(
- new ClusterCompactionConfig(0.1, 9, true, null),
+ new ClusterCompactionConfig(0.1, 9, null, null, null),
DataSourceCompactionConfig.builder().forDataSource(TestDataSource.WIKI).build(),
auditInfo,
DateTimes.nowUtc()
@@ -43,7 +43,7 @@ public class DataSourceCompactionConfigAuditEntryTest
public void testhasSameConfigWithSameBaseConfigIsTrue()
{
final DataSourceCompactionConfigAuditEntry secondEntry = new DataSourceCompactionConfigAuditEntry(
- new ClusterCompactionConfig(0.1, 9, true, null),
+ new ClusterCompactionConfig(0.1, 9, null, null, null),
DataSourceCompactionConfig.builder().forDataSource(TestDataSource.WIKI).build(),
auditInfo,
DateTimes.nowUtc()
@@ -56,7 +56,7 @@ public class DataSourceCompactionConfigAuditEntryTest
public void testhasSameConfigWithDifferentClusterConfigIsFalse()
{
DataSourceCompactionConfigAuditEntry secondEntry = new DataSourceCompactionConfigAuditEntry(
- new ClusterCompactionConfig(0.2, 9, false, null),
+ new ClusterCompactionConfig(0.2, 9, null, null, null),
DataSourceCompactionConfig.builder().forDataSource(TestDataSource.WIKI).build(),
auditInfo,
DateTimes.nowUtc()
@@ -65,7 +65,7 @@ public class DataSourceCompactionConfigAuditEntryTest
Assert.assertFalse(secondEntry.hasSameConfig(firstEntry));
secondEntry = new DataSourceCompactionConfigAuditEntry(
- new ClusterCompactionConfig(0.1, 10, true, null),
+ new ClusterCompactionConfig(0.1, 10, null, null, null),
DataSourceCompactionConfig.builder().forDataSource(TestDataSource.WIKI).build(),
auditInfo,
DateTimes.nowUtc()
@@ -78,7 +78,7 @@ public class DataSourceCompactionConfigAuditEntryTest
public void testhasSameConfigWithDifferentDatasourceConfigIsFalse()
{
DataSourceCompactionConfigAuditEntry secondEntry = new DataSourceCompactionConfigAuditEntry(
- new ClusterCompactionConfig(0.1, 9, true, null),
+ new ClusterCompactionConfig(0.1, 9, null, null, null),
DataSourceCompactionConfig.builder().forDataSource(TestDataSource.KOALA).build(),
auditInfo,
DateTimes.nowUtc()
@@ -91,7 +91,7 @@ public class DataSourceCompactionConfigAuditEntryTest
public void testhasSameConfigWithNullDatasourceConfigIsFalse()
{
final DataSourceCompactionConfigAuditEntry secondEntry = new DataSourceCompactionConfigAuditEntry(
- new ClusterCompactionConfig(0.1, 9, true, null),
+ new ClusterCompactionConfig(0.1, 9, null, null, null),
null,
auditInfo,
DateTimes.nowUtc()
diff --git a/server/src/test/java/org/apache/druid/server/coordinator/DataSourceCompactionConfigHistoryTest.java b/server/src/test/java/org/apache/druid/server/coordinator/DataSourceCompactionConfigHistoryTest.java
index bcdee91ecd5..ba670c31248 100644
--- a/server/src/test/java/org/apache/druid/server/coordinator/DataSourceCompactionConfigHistoryTest.java
+++ b/server/src/test/java/org/apache/druid/server/coordinator/DataSourceCompactionConfigHistoryTest.java
@@ -177,7 +177,7 @@ public class DataSourceCompactionConfigHistoryTest
wikiAuditHistory.add(originalConfig, auditInfo, DateTimes.nowUtc());
final DruidCompactionConfig updatedConfig = originalConfig.withClusterConfig(
- new ClusterCompactionConfig(0.2, null, null, null)
+ new ClusterCompactionConfig(0.2, null, null, null, null)
);
wikiAuditHistory.add(updatedConfig, auditInfo, DateTimes.nowUtc());
diff --git a/server/src/test/java/org/apache/druid/server/coordinator/DruidCompactionConfigTest.java b/server/src/test/java/org/apache/druid/server/coordinator/DruidCompactionConfigTest.java
index 2f481727fa1..ffd3b453b8f 100644
--- a/server/src/test/java/org/apache/druid/server/coordinator/DruidCompactionConfigTest.java
+++ b/server/src/test/java/org/apache/druid/server/coordinator/DruidCompactionConfigTest.java
@@ -64,6 +64,7 @@ public class DruidCompactionConfigTest
null,
null,
null,
+ null,
null
);
@@ -80,8 +81,9 @@ public class DruidCompactionConfigTest
final ClusterCompactionConfig clusterConfig = new ClusterCompactionConfig(
0.5,
10,
- false,
- new NewestSegmentFirstPolicy(null)
+ new NewestSegmentFirstPolicy(null),
+ true,
+ CompactionEngine.MSQ
);
final DruidCompactionConfig copy = config.withClusterConfig(clusterConfig);
@@ -106,4 +108,15 @@ public class DruidCompactionConfigTest
Assert.assertEquals(1, copy.getCompactionConfigs().size());
Assert.assertEquals(dataSourceConfig, copy.findConfigForDatasource(TestDataSource.WIKI).orNull());
}
+
+ @Test
+ public void testDefaultConfigValues()
+ {
+ final DruidCompactionConfig config = DruidCompactionConfig.empty();
+ Assert.assertTrue(config.getCompactionConfigs().isEmpty());
+ Assert.assertTrue(config.getCompactionPolicy() instanceof NewestSegmentFirstPolicy);
+ Assert.assertEquals(CompactionEngine.NATIVE, config.getEngine());
+ Assert.assertEquals(0.1, config.getCompactionTaskSlotRatio(), 1e-9);
+ Assert.assertEquals(Integer.MAX_VALUE, config.getMaxCompactionTaskSlots());
+ }
}
diff --git a/server/src/test/java/org/apache/druid/server/coordinator/DruidCoordinatorTest.java b/server/src/test/java/org/apache/druid/server/coordinator/DruidCoordinatorTest.java
index d50e82e7baa..abe5dce051f 100644
--- a/server/src/test/java/org/apache/druid/server/coordinator/DruidCoordinatorTest.java
+++ b/server/src/test/java/org/apache/druid/server/coordinator/DruidCoordinatorTest.java
@@ -821,7 +821,7 @@ public class DruidCoordinatorTest
.anyTimes();
EasyMock.replay(segmentsMetadataManager);
CompactionSimulateResult result = coordinator.simulateRunWithConfigUpdate(
- new ClusterCompactionConfig(0.2, null, null, null)
+ new ClusterCompactionConfig(0.2, null, null, null, null)
);
Assert.assertEquals(Collections.emptyMap(), result.getCompactionStates());
}
diff --git a/server/src/test/java/org/apache/druid/server/coordinator/duty/CompactSegmentsTest.java b/server/src/test/java/org/apache/druid/server/coordinator/duty/CompactSegmentsTest.java
index e1a5ed52e33..3a887e9fec3 100644
--- a/server/src/test/java/org/apache/druid/server/coordinator/duty/CompactSegmentsTest.java
+++ b/server/src/test/java/org/apache/druid/server/coordinator/duty/CompactSegmentsTest.java
@@ -672,7 +672,7 @@ public class CompactSegmentsTest
final TestOverlordClient overlordClient = new TestOverlordClient(JSON_MAPPER);
final CompactSegments compactSegments = new CompactSegments(statusTracker, overlordClient);
final CoordinatorRunStats stats =
- doCompactSegments(compactSegments, createCompactionConfigs(), maxCompactionSlot, true);
+ doCompactSegments(compactSegments, createCompactionConfigs(), maxCompactionSlot);
Assert.assertEquals(maxCompactionSlot, stats.get(Stats.Compaction.AVAILABLE_SLOTS));
Assert.assertEquals(maxCompactionSlot, stats.get(Stats.Compaction.MAX_SLOTS));
// Native takes up 1 task slot by default whereas MSQ takes up all available upto 5. Since there are 3 available
@@ -692,7 +692,7 @@ public class CompactSegmentsTest
final TestOverlordClient overlordClient = new TestOverlordClient(JSON_MAPPER);
final CompactSegments compactSegments = new CompactSegments(statusTracker, overlordClient);
final CoordinatorRunStats stats =
- doCompactSegments(compactSegments, createCompactionConfigs(), maxCompactionSlot, true);
+ doCompactSegments(compactSegments, createCompactionConfigs(), maxCompactionSlot);
Assert.assertEquals(MAXIMUM_CAPACITY_WITH_AUTO_SCALE, stats.get(Stats.Compaction.AVAILABLE_SLOTS));
Assert.assertEquals(MAXIMUM_CAPACITY_WITH_AUTO_SCALE, stats.get(Stats.Compaction.MAX_SLOTS));
// Native takes up 1 task slot by default whereas MSQ takes up all available upto 5. Since there are 10 available
@@ -1865,16 +1865,6 @@ public class CompactSegmentsTest
List<DataSourceCompactionConfig> compactionConfigs,
@Nullable Integer numCompactionTaskSlots
)
- {
- return doCompactSegments(compactSegments, compactionConfigs, numCompactionTaskSlots, false);
- }
-
- private CoordinatorRunStats doCompactSegments(
- CompactSegments compactSegments,
- List<DataSourceCompactionConfig> compactionConfigs,
- @Nullable Integer numCompactionTaskSlots,
- boolean useAutoScaleSlots
- )
{
DruidCoordinatorRuntimeParams params = DruidCoordinatorRuntimeParams
.builder()
@@ -1884,7 +1874,8 @@ public class CompactSegmentsTest
compactionConfigs,
numCompactionTaskSlots == null ? null : 1.0, // 100% when numCompactionTaskSlots is not null
numCompactionTaskSlots,
- useAutoScaleSlots,
+ null,
+ null,
null
)
)
diff --git a/server/src/test/java/org/apache/druid/server/http/CoordinatorCompactionConfigsResourceTest.java b/server/src/test/java/org/apache/druid/server/http/CoordinatorCompactionConfigsResourceTest.java
index 4cef66d2ac5..029c0290c91 100644
--- a/server/src/test/java/org/apache/druid/server/http/CoordinatorCompactionConfigsResourceTest.java
+++ b/server/src/test/java/org/apache/druid/server/http/CoordinatorCompactionConfigsResourceTest.java
@@ -104,15 +104,16 @@ public class CoordinatorCompactionConfigsResourceTest
Assert.assertEquals(0.1, defaultConfig.getCompactionTaskSlotRatio(), DELTA);
Assert.assertEquals(Integer.MAX_VALUE, defaultConfig.getMaxCompactionTaskSlots());
- Assert.assertFalse(defaultConfig.isUseAutoScaleSlots());
Assert.assertTrue(defaultConfig.getCompactionConfigs().isEmpty());
+ Assert.assertFalse(defaultConfig.isUseSupervisors());
+ Assert.assertEquals(CompactionEngine.NATIVE, defaultConfig.getEngine());
}
@Test
public void testUpdateClusterConfig()
{
Response response = resource.updateClusterCompactionConfig(
- new ClusterCompactionConfig(0.5, 10, true, null),
+ new ClusterCompactionConfig(0.5, 10, null, true, CompactionEngine.MSQ),
mockHttpServletRequest
);
verifyStatus(Response.Status.OK, response);
@@ -125,7 +126,8 @@ public class CoordinatorCompactionConfigsResourceTest
Assert.assertNotNull(updatedConfig);
Assert.assertEquals(0.5, updatedConfig.getCompactionTaskSlotRatio(), DELTA);
Assert.assertEquals(10, updatedConfig.getMaxCompactionTaskSlots());
- Assert.assertTrue(updatedConfig.isUseAutoScaleSlots());
+ Assert.assertTrue(updatedConfig.isUseSupervisors());
+ Assert.assertEquals(CompactionEngine.MSQ, updatedConfig.getEngine());
}
@Test
@@ -134,7 +136,7 @@ public class CoordinatorCompactionConfigsResourceTest
final DruidCompactionConfig defaultConfig
= verifyAndGetPayload(resource.getCompactionConfig(), DruidCompactionConfig.class);
- Response response = resource.setCompactionTaskLimit(0.5, 9, true, mockHttpServletRequest);
+ Response response = resource.setCompactionTaskLimit(0.5, 9, mockHttpServletRequest);
verifyStatus(Response.Status.OK, response);
final DruidCompactionConfig updatedConfig
@@ -143,7 +145,6 @@ public class CoordinatorCompactionConfigsResourceTest
// Verify that the task slot fields have been updated
Assert.assertEquals(0.5, updatedConfig.getCompactionTaskSlotRatio(), DELTA);
Assert.assertEquals(9, updatedConfig.getMaxCompactionTaskSlots());
- Assert.assertTrue(updatedConfig.isUseAutoScaleSlots());
// Verify that the other fields are unchanged
Assert.assertEquals(defaultConfig.getCompactionConfigs(), updatedConfig.getCompactionConfigs());
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org