getDataSources()
{
diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/compact/InlineCompactionJobTemplate.java b/indexing-service/src/main/java/org/apache/druid/indexing/compact/InlineCompactionJobTemplate.java
new file mode 100644
index 000000000000..c147fde4f9f2
--- /dev/null
+++ b/indexing-service/src/main/java/org/apache/druid/indexing/compact/InlineCompactionJobTemplate.java
@@ -0,0 +1,117 @@
+/*
+ * 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.indexing.compact;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.druid.data.input.InputSource;
+import org.apache.druid.data.output.OutputDestination;
+import org.apache.druid.java.util.common.granularity.Granularity;
+import org.apache.druid.server.coordinator.InlineSchemaDataSourceCompactionConfig;
+import org.apache.druid.server.coordinator.UserCompactionTaskGranularityConfig;
+import org.apache.druid.server.coordinator.UserCompactionTaskQueryTuningConfig;
+import org.joda.time.Period;
+
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * Template to create compaction jobs using inline specifications. This template
+ * does not fetch any information from the Druid catalog while creating jobs.
+ *
+ * This template does not contain all the fields supported by
+ * {@link InlineSchemaDataSourceCompactionConfig} since some of those fields may
+ * change the data itself (and not just its layout) and are thus not considered
+ * compaction-compatible.
+ */
+public class InlineCompactionJobTemplate extends CompactionJobTemplate
+{
+ public static final String TYPE = "compactInline";
+
+ private final UserCompactionTaskQueryTuningConfig tuningConfig;
+ private final Granularity segmentGranularity;
+
+ @JsonCreator
+ public InlineCompactionJobTemplate(
+ @JsonProperty("tuningConfig") UserCompactionTaskQueryTuningConfig tuningConfig,
+ @JsonProperty("segmentGranularity") Granularity segmentGranularity
+ )
+ {
+ this.tuningConfig = tuningConfig;
+ this.segmentGranularity = segmentGranularity;
+ }
+
+ @JsonProperty
+ public Granularity getSegmentGranularity()
+ {
+ return segmentGranularity;
+ }
+
+ @JsonProperty
+ public UserCompactionTaskQueryTuningConfig getTuningConfig()
+ {
+ return tuningConfig;
+ }
+
+ @Override
+ public List createCompactionJobs(
+ InputSource source,
+ OutputDestination destination,
+ CompactionJobParams jobParams
+ )
+ {
+ final String dataSource = ensureDruidInputSource(source).getDataSource();
+ return new CompactionConfigBasedJobTemplate(
+ InlineSchemaDataSourceCompactionConfig
+ .builder()
+ .forDataSource(dataSource)
+ .withSkipOffsetFromLatest(Period.ZERO)
+ .withTuningConfig(tuningConfig)
+ .withGranularitySpec(new UserCompactionTaskGranularityConfig(segmentGranularity, null, null))
+ .build()
+ ).createCompactionJobs(source, destination, jobParams);
+ }
+
+ @Override
+ public boolean equals(Object object)
+ {
+ if (this == object) {
+ return true;
+ }
+ if (object == null || getClass() != object.getClass()) {
+ return false;
+ }
+ InlineCompactionJobTemplate that = (InlineCompactionJobTemplate) object;
+ return Objects.equals(this.tuningConfig, that.tuningConfig)
+ && Objects.equals(this.segmentGranularity, that.segmentGranularity);
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(tuningConfig, segmentGranularity);
+ }
+
+ @Override
+ public String getType()
+ {
+ return TYPE;
+ }
+}
diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/compact/LocalOverlordClient.java b/indexing-service/src/main/java/org/apache/druid/indexing/compact/LocalOverlordClient.java
index 3f1427c7c34e..eb200577bb8b 100644
--- a/indexing-service/src/main/java/org/apache/druid/indexing/compact/LocalOverlordClient.java
+++ b/indexing-service/src/main/java/org/apache/druid/indexing/compact/LocalOverlordClient.java
@@ -72,10 +72,12 @@ class LocalOverlordClient extends NoopOverlordClient
@Override
public ListenableFuture runTask(String taskId, Object clientTaskQuery)
{
+ final CompactionTask task =
+ clientTaskQuery instanceof CompactionTask
+ ? (CompactionTask) clientTaskQuery
+ : convertTask(clientTaskQuery, ClientCompactionTaskQuery.class, CompactionTask.class);
return futureOf(() -> {
- getValidTaskQueue().add(
- convertTask(clientTaskQuery, ClientCompactionTaskQuery.class, CompactionTask.class)
- );
+ getValidTaskQueue().add(task);
return null;
});
}
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 95ea3d116939..9896db489264 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
@@ -27,6 +27,8 @@
import org.apache.druid.client.indexing.ClientCompactionRunnerInfo;
import org.apache.druid.indexer.TaskLocation;
import org.apache.druid.indexer.TaskStatus;
+import org.apache.druid.indexing.common.actions.TaskActionClientFactory;
+import org.apache.druid.indexing.overlord.GlobalTaskLockbox;
import org.apache.druid.indexing.overlord.TaskMaster;
import org.apache.druid.indexing.overlord.TaskQueryTool;
import org.apache.druid.indexing.overlord.TaskRunner;
@@ -52,17 +54,19 @@
import org.apache.druid.server.coordinator.duty.CompactSegments;
import org.apache.druid.server.coordinator.stats.CoordinatorRunStats;
import org.apache.druid.server.coordinator.stats.CoordinatorStat;
-import org.apache.druid.server.coordinator.stats.Dimension;
+import org.apache.druid.server.coordinator.stats.RowKey;
import org.apache.druid.server.coordinator.stats.Stats;
import org.joda.time.Duration;
-import java.util.ArrayList;
import java.util.Collections;
+import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
/**
* Implementation of {@link CompactionScheduler}.
@@ -87,10 +91,13 @@ public class OverlordCompactionScheduler implements CompactionScheduler
private final SegmentsMetadataManager segmentManager;
private final LocalOverlordClient overlordClient;
private final ServiceEmitter emitter;
+ private final ObjectMapper objectMapper;
private final TaskMaster taskMaster;
private final Supplier compactionConfigSupplier;
- private final ConcurrentHashMap activeDatasourceConfigs;
+ private final ConcurrentHashMap activeSupervisors;
+
+ private final AtomicReference> datasourceToCompactionSnapshot;
/**
* Single-threaded executor to process the compaction queue.
@@ -98,6 +105,8 @@ public class OverlordCompactionScheduler implements CompactionScheduler
private final ScheduledExecutorService executor;
private final CompactionStatusTracker statusTracker;
+ private final TaskActionClientFactory taskActionClientFactory;
+ private final GlobalTaskLockbox taskLockbox;
/**
* Listener to watch task completion events and update CompactionStatusTracker.
@@ -107,7 +116,6 @@ public class OverlordCompactionScheduler implements CompactionScheduler
private final AtomicBoolean isLeader = new AtomicBoolean(false);
private final AtomicBoolean started = new AtomicBoolean(false);
- private final CompactSegments duty;
/**
* The scheduler should enable/disable polling of segments only if the Overlord
@@ -121,11 +129,13 @@ public class OverlordCompactionScheduler implements CompactionScheduler
@Inject
public OverlordCompactionScheduler(
TaskMaster taskMaster,
+ GlobalTaskLockbox taskLockbox,
TaskQueryTool taskQueryTool,
SegmentsMetadataManager segmentManager,
Supplier compactionConfigSupplier,
CompactionStatusTracker statusTracker,
CoordinatorOverlordServiceConfig coordinatorOverlordServiceConfig,
+ TaskActionClientFactory taskActionClientFactory,
ScheduledExecutorFactory executorFactory,
ServiceEmitter emitter,
ObjectMapper objectMapper
@@ -133,7 +143,9 @@ public OverlordCompactionScheduler(
{
this.segmentManager = segmentManager;
this.emitter = emitter;
+ this.objectMapper = objectMapper;
this.taskMaster = taskMaster;
+ this.taskLockbox = taskLockbox;
this.compactionConfigSupplier = compactionConfigSupplier;
this.executor = executorFactory.create(1, "CompactionScheduler-%s");
@@ -141,9 +153,10 @@ public OverlordCompactionScheduler(
this.shouldPollSegments = segmentManager != null
&& !coordinatorOverlordServiceConfig.isEnabled();
this.overlordClient = new LocalOverlordClient(taskMaster, taskQueryTool, objectMapper);
- this.duty = new CompactSegments(this.statusTracker, overlordClient);
- this.activeDatasourceConfigs = new ConcurrentHashMap<>();
+ this.activeSupervisors = new ConcurrentHashMap<>();
+ this.datasourceToCompactionSnapshot = new AtomicReference<>();
+ this.taskActionClientFactory = taskActionClientFactory;
this.taskRunnerListener = new TaskRunnerListener()
{
@Override
@@ -208,25 +221,25 @@ public CompactionConfigValidationResult validateCompactionConfig(DataSourceCompa
} else {
return ClientCompactionRunnerInfo.validateCompactionConfig(
compactionConfig,
- getLatestConfig().getEngine()
+ getLatestClusterConfig().getEngine()
);
}
}
@Override
- public void startCompaction(String dataSourceName, DataSourceCompactionConfig config)
+ public void startCompaction(String dataSourceName, CompactionSupervisor supervisor)
{
// Track active datasources even if scheduler has not started yet because
// SupervisorManager is started before the scheduler
if (isEnabled()) {
- activeDatasourceConfigs.put(dataSourceName, config);
+ activeSupervisors.put(dataSourceName, supervisor);
}
}
@Override
public void stopCompaction(String dataSourceName)
{
- activeDatasourceConfigs.remove(dataSourceName);
+ activeSupervisors.remove(dataSourceName);
statusTracker.removeDatasource(dataSourceName);
}
@@ -264,7 +277,7 @@ private synchronized void cleanupState()
taskRunnerOptional.get().unregisterListener(taskRunnerListener.getListenerId());
}
statusTracker.stop();
- activeDatasourceConfigs.clear();
+ activeSupervisors.clear();
if (shouldPollSegments) {
segmentManager.stopPollingDatabasePeriodically();
@@ -304,41 +317,52 @@ private synchronized void scheduledRun()
}
/**
- * Runs the compaction duty and emits stats if {@link #METRIC_EMISSION_PERIOD}
- * has elapsed.
+ * Creates and runs eligible compaction jobs.
*/
private synchronized void runCompactionDuty()
{
- final CoordinatorRunStats stats = new CoordinatorRunStats();
- duty.run(getLatestConfig(), getDatasourceSnapshot(), getLatestConfig().getEngine(), stats);
+ final CompactionJobQueue queue = new CompactionJobQueue(
+ getDatasourceSnapshot(),
+ getLatestClusterConfig(),
+ statusTracker,
+ taskActionClientFactory,
+ taskLockbox,
+ overlordClient,
+ objectMapper
+ );
+ statusTracker.resetActiveDatasources(activeSupervisors.keySet());
+ activeSupervisors.forEach((datasource, supervisor) -> queue.createAndEnqueueJobs(supervisor));
+ queue.runReadyJobs();
+
+ datasourceToCompactionSnapshot.set(queue.getCompactionSnapshots());
+ emitStatsIfPeriodHasElapsed(queue.getRunStats());
+ }
+ /**
+ * Emits stats if {@link #METRIC_EMISSION_PERIOD} has elapsed.
+ */
+ private void emitStatsIfPeriodHasElapsed(CoordinatorRunStats stats)
+ {
// Emit stats only if emission period has elapsed
if (!sinceStatsEmitted.isRunning() || sinceStatsEmitted.hasElapsed(METRIC_EMISSION_PERIOD)) {
- stats.forEachStat(
- (stat, dimensions, value) -> {
- if (stat.shouldEmit()) {
- emitStat(stat, dimensions.getValues(), value);
- }
- }
- );
+ stats.forEachStat(this::emitStat);
sinceStatsEmitted.restart();
} else {
// Always emit number of submitted tasks
- long numSubmittedTasks = stats.get(Stats.Compaction.SUBMITTED_TASKS);
- emitStat(Stats.Compaction.SUBMITTED_TASKS, Collections.emptyMap(), numSubmittedTasks);
+ stats.forEachEntry(Stats.Compaction.SUBMITTED_TASKS, this::emitStat);
}
}
@Override
public AutoCompactionSnapshot getCompactionSnapshot(String dataSource)
{
- if (!activeDatasourceConfigs.containsKey(dataSource)) {
+ if (!activeSupervisors.containsKey(dataSource)) {
return AutoCompactionSnapshot.builder(dataSource)
.withStatus(AutoCompactionSnapshot.ScheduleStatus.NOT_ENABLED)
.build();
}
- final AutoCompactionSnapshot snapshot = duty.getAutoCompactionSnapshot(dataSource);
+ final AutoCompactionSnapshot snapshot = datasourceToCompactionSnapshot.get().get(dataSource);
if (snapshot == null) {
final AutoCompactionSnapshot.ScheduleStatus status =
isEnabled()
@@ -353,7 +377,7 @@ public AutoCompactionSnapshot getCompactionSnapshot(String dataSource)
@Override
public Map getAllCompactionSnapshots()
{
- return duty.getAutoCompactionSnapshot();
+ return Map.copyOf(datasourceToCompactionSnapshot.get());
}
@Override
@@ -363,17 +387,21 @@ public CompactionSimulateResult simulateRunWithConfigUpdate(ClusterCompactionCon
return new CompactionRunSimulator(statusTracker, overlordClient).simulateRunWithConfig(
getLatestConfig().withClusterConfig(updateRequest),
getDatasourceSnapshot(),
- getLatestConfig().getEngine()
+ updateRequest.getEngine()
);
} else {
return new CompactionSimulateResult(Collections.emptyMap());
}
}
- private void emitStat(CoordinatorStat stat, Map dimensionValues, long value)
+ private void emitStat(CoordinatorStat stat, RowKey rowKey, long value)
{
+ if (!stat.shouldEmit()) {
+ return;
+ }
+
ServiceMetricEvent.Builder eventBuilder = new ServiceMetricEvent.Builder();
- dimensionValues.forEach(
+ rowKey.getValues().forEach(
(dim, dimValue) -> eventBuilder.setDimension(dim.reportedName(), dimValue)
);
emitter.emit(eventBuilder.setMetric(stat.getMetricName(), value));
@@ -381,10 +409,20 @@ private void emitStat(CoordinatorStat stat, Map dimensionValu
private DruidCompactionConfig getLatestConfig()
{
+ final List configs = activeSupervisors
+ .values()
+ .stream()
+ .map(s -> s.getSpec().getSpec())
+ .collect(Collectors.toList());
return DruidCompactionConfig
.empty()
- .withClusterConfig(compactionConfigSupplier.get().clusterConfig())
- .withDatasourceConfigs(new ArrayList<>(activeDatasourceConfigs.values()));
+ .withClusterConfig(getLatestClusterConfig())
+ .withDatasourceConfigs(configs);
+ }
+
+ private ClusterCompactionConfig getLatestClusterConfig()
+ {
+ return compactionConfigSupplier.get().clusterConfig();
}
private DataSourcesSnapshot getDatasourceSnapshot()
diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/input/DruidDatasourceDestination.java b/indexing-service/src/main/java/org/apache/druid/indexing/input/DruidDatasourceDestination.java
new file mode 100644
index 000000000000..7d10ce86bd75
--- /dev/null
+++ b/indexing-service/src/main/java/org/apache/druid/indexing/input/DruidDatasourceDestination.java
@@ -0,0 +1,42 @@
+/*
+ * 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.indexing.input;
+
+import org.apache.druid.data.output.OutputDestination;
+
+/**
+ * {@link OutputDestination} for writing out data into a Druid datasource.
+ */
+public class DruidDatasourceDestination implements OutputDestination
+{
+ public static final String TYPE = "druid";
+
+ private final String dataSource;
+
+ public DruidDatasourceDestination(String dataSource)
+ {
+ this.dataSource = dataSource;
+ }
+
+ public String getDataSource()
+ {
+ return dataSource;
+ }
+}
diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/input/DruidInputSource.java b/indexing-service/src/main/java/org/apache/druid/indexing/input/DruidInputSource.java
index 84c17f1a8fa0..fea6f30ef444 100644
--- a/indexing-service/src/main/java/org/apache/druid/indexing/input/DruidInputSource.java
+++ b/indexing-service/src/main/java/org/apache/druid/indexing/input/DruidInputSource.java
@@ -291,6 +291,25 @@ public InputSource withTaskToolbox(TaskToolbox toolbox)
);
}
+ /**
+ * Creates a new {@link DruidInputSource} with the given interval.
+ */
+ public DruidInputSource withInterval(Interval interval)
+ {
+ return new DruidInputSource(
+ this.dataSource,
+ interval,
+ null,
+ this.dimFilter,
+ this.dimensions,
+ this.metrics,
+ this.indexIO,
+ this.coordinatorClient,
+ this.segmentCacheManagerFactory,
+ this.taskConfig
+ );
+ }
+
@Override
protected InputSourceReader fixedFormatReader(InputRowSchema inputRowSchema, @Nullable File temporaryDirectory)
{
diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/BatchIndexingSupervisor.java b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/BatchIndexingSupervisor.java
new file mode 100644
index 000000000000..c26c27fc1038
--- /dev/null
+++ b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/BatchIndexingSupervisor.java
@@ -0,0 +1,48 @@
+/*
+ * 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.indexing.overlord.supervisor;
+
+import org.apache.druid.indexing.template.BatchIndexingJob;
+import org.apache.druid.indexing.template.JobParams;
+
+import java.util.List;
+
+/**
+ * Supervisor to perform batch ingestion using {@link BatchIndexingJob}.
+ */
+public interface BatchIndexingSupervisor
+ extends Supervisor
+{
+ /**
+ * Checks if this supervisor is ready to create jobs in the current run.
+ *
+ * @param jobParams Parameters for the current run of the scheduler.
+ */
+ boolean shouldCreateJobs(P jobParams);
+
+ /**
+ * Creates jobs to be launched in the current run of the scheduler.
+ *
+ * @param jobParams Parameters for the current run of the scheduler.
+ * @return Empty iterator if no tasks are to be submitted in the current run
+ * of the scheduler.
+ */
+ List createJobs(P jobParams);
+}
diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/BatchIndexingSupervisorSpec.java b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/BatchIndexingSupervisorSpec.java
new file mode 100644
index 000000000000..30e2f2b5109c
--- /dev/null
+++ b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/BatchIndexingSupervisorSpec.java
@@ -0,0 +1,41 @@
+/*
+ * 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.indexing.overlord.supervisor;
+
+import org.apache.druid.indexing.template.BatchIndexingJob;
+import org.apache.druid.indexing.template.BatchIndexingJobTemplate;
+import org.apache.druid.indexing.template.JobParams;
+
+/**
+ * Spec for {@link BatchIndexingSupervisor}. Provides a template to create
+ * {@link BatchIndexingJob}.
+ */
+public interface BatchIndexingSupervisorSpec
+
+ extends SupervisorSpec
+{
+ @Override
+ BatchIndexingSupervisor createSupervisor();
+
+ /**
+ * Template used by the corresponding supervisor to create {@link BatchIndexingJob}s.
+ */
+ BatchIndexingJobTemplate getTemplate();
+}
diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/compact/CompactionSupervisorSpecTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/compact/CompactionSupervisorSpecTest.java
index b1732acabbbc..b008d74829ae 100644
--- a/indexing-service/src/test/java/org/apache/druid/indexing/compact/CompactionSupervisorSpecTest.java
+++ b/indexing-service/src/test/java/org/apache/druid/indexing/compact/CompactionSupervisorSpecTest.java
@@ -137,7 +137,7 @@ public void testStartStopSupervisorForActiveSpec()
supervisor.start();
supervisor.stop(false);
- Mockito.verify(scheduler, Mockito.times(1)).startCompaction(TestDataSource.WIKI, spec);
+ Mockito.verify(scheduler, Mockito.times(1)).startCompaction(TestDataSource.WIKI, supervisor);
Mockito.verify(scheduler, Mockito.times(1)).stopCompaction(TestDataSource.WIKI);
}
@@ -157,7 +157,7 @@ public void testStartStopSupervisorWhenSchedulerStopped()
supervisor.start();
supervisor.stop(false);
- Mockito.verify(scheduler, Mockito.times(1)).startCompaction(TestDataSource.WIKI, spec);
+ Mockito.verify(scheduler, Mockito.times(1)).startCompaction(TestDataSource.WIKI, supervisor);
Mockito.verify(scheduler, Mockito.times(1)).stopCompaction(TestDataSource.WIKI);
}
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 9353953fc2af..86bbf3dff614 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
@@ -21,10 +21,20 @@
import com.fasterxml.jackson.databind.InjectableValues;
import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.druid.client.coordinator.CoordinatorClient;
+import org.apache.druid.client.coordinator.NoopCoordinatorClient;
import org.apache.druid.client.indexing.ClientMSQContext;
+import org.apache.druid.guice.IndexingServiceInputSourceModule;
import org.apache.druid.guice.IndexingServiceTuningConfigModule;
import org.apache.druid.indexer.CompactionEngine;
import org.apache.druid.indexing.common.SegmentCacheManagerFactory;
+import org.apache.druid.indexing.common.TimeChunkLock;
+import org.apache.druid.indexing.common.actions.RetrieveUsedSegmentsAction;
+import org.apache.druid.indexing.common.actions.TaskAction;
+import org.apache.druid.indexing.common.actions.TaskActionClient;
+import org.apache.druid.indexing.common.actions.TaskActionClientFactory;
+import org.apache.druid.indexing.common.actions.TimeChunkLockTryAcquireAction;
+import org.apache.druid.indexing.common.config.TaskConfig;
import org.apache.druid.indexing.common.config.TaskStorageConfig;
import org.apache.druid.indexing.common.task.CompactionTask;
import org.apache.druid.indexing.common.task.Task;
@@ -38,9 +48,11 @@
import org.apache.druid.indexing.overlord.setup.WorkerBehaviorConfig;
import org.apache.druid.indexing.test.TestIndexerMetadataStorageCoordinator;
import org.apache.druid.jackson.DefaultObjectMapper;
+import org.apache.druid.java.util.common.DateTimes;
import org.apache.druid.java.util.common.Intervals;
import org.apache.druid.java.util.common.granularity.Granularities;
import org.apache.druid.java.util.metrics.StubServiceEmitter;
+import org.apache.druid.segment.IndexIO;
import org.apache.druid.segment.TestDataSource;
import org.apache.druid.segment.TestIndex;
import org.apache.druid.server.compaction.CompactionSimulateResult;
@@ -81,9 +93,13 @@ public class OverlordCompactionSchedulerTest
static {
OBJECT_MAPPER = new DefaultObjectMapper();
OBJECT_MAPPER.registerModules(new IndexingServiceTuningConfigModule().getJacksonModules());
+ OBJECT_MAPPER.registerModules(new IndexingServiceInputSourceModule().getJacksonModules());
OBJECT_MAPPER.setInjectableValues(
new InjectableValues
.Std()
+ .addValue(IndexIO.class, TestIndex.INDEX_IO)
+ .addValue(TaskConfig.class, Mockito.mock(TaskConfig.class))
+ .addValue(CoordinatorClient.class, new NoopCoordinatorClient())
.addValue(
SegmentCacheManagerFactory.class,
new SegmentCacheManagerFactory(TestIndex.INDEX_IO, OBJECT_MAPPER)
@@ -96,6 +112,7 @@ public class OverlordCompactionSchedulerTest
private TaskMaster taskMaster;
private TaskQueue taskQueue;
+ private TaskActionClientFactory taskActionClientFactory;
private BlockingExecutorService executor;
private HeapMemoryTaskStorage taskStorage;
@@ -131,6 +148,30 @@ public void setUp()
compactionConfig = new AtomicReference<>(new ClusterCompactionConfig(null, null, null, true, null));
coordinatorOverlordServiceConfig = new CoordinatorOverlordServiceConfig(false, null);
+ taskActionClientFactory = task -> new TaskActionClient()
+ {
+ @Override
+ @SuppressWarnings("unchecked")
+ public RetType submit(TaskAction taskAction)
+ {
+ if (taskAction instanceof RetrieveUsedSegmentsAction) {
+ return (RetType) segmentsMetadataManager.getAllSegments();
+ } else if (taskAction instanceof TimeChunkLockTryAcquireAction) {
+ final TimeChunkLockTryAcquireAction lockAcquireAction = (TimeChunkLockTryAcquireAction) taskAction;
+ return (RetType) new TimeChunkLock(
+ null,
+ task.getGroupId(),
+ task.getDataSource(),
+ lockAcquireAction.getInterval(),
+ DateTimes.nowUtc().toString(),
+ 1
+ );
+ } else {
+ return null;
+ }
+ }
+ };
+
initScheduler();
}
@@ -142,11 +183,13 @@ private void initScheduler()
= new DefaultWorkerBehaviorConfig(WorkerBehaviorConfig.DEFAULT_STRATEGY, null);
scheduler = new OverlordCompactionScheduler(
taskMaster,
+ taskLockbox,
new TaskQueryTool(taskStorage, taskLockbox, taskMaster, null, () -> defaultWorkerConfig),
segmentsMetadataManager,
() -> DruidCompactionConfig.empty().withClusterConfig(compactionConfig.get()),
- new CompactionStatusTracker(OBJECT_MAPPER),
+ new CompactionStatusTracker(),
coordinatorOverlordServiceConfig,
+ taskActionClientFactory,
(nameFormat, numThreads) -> new WrappingScheduledExecutorService("test", executor, false),
serviceEmitter,
OBJECT_MAPPER
@@ -304,13 +347,7 @@ public void testStartCompaction()
wikiSegments.forEach(segmentsMetadataManager::addSegment);
scheduler.becomeLeader();
- scheduler.startCompaction(
- TestDataSource.WIKI,
- InlineSchemaDataSourceCompactionConfig.builder()
- .forDataSource(TestDataSource.WIKI)
- .withSkipOffsetFromLatest(Period.seconds(0))
- .build()
- );
+ scheduler.startCompaction(TestDataSource.WIKI, createSupervisor());
executor.finishNextPendingTask();
@@ -351,10 +388,7 @@ public void testStopCompaction()
scheduler.becomeLeader();
scheduler.startCompaction(
TestDataSource.WIKI,
- InlineSchemaDataSourceCompactionConfig.builder()
- .forDataSource(TestDataSource.WIKI)
- .withSkipOffsetFromLatest(Period.seconds(0))
- .build()
+ createSupervisor()
);
scheduler.stopCompaction(TestDataSource.WIKI);
@@ -390,13 +424,7 @@ public void testSimulateRun()
scheduler.becomeLeader();
runScheduledJob();
- scheduler.startCompaction(
- TestDataSource.WIKI,
- InlineSchemaDataSourceCompactionConfig.builder()
- .forDataSource(TestDataSource.WIKI)
- .withSkipOffsetFromLatest(Period.seconds(0))
- .build()
- );
+ scheduler.startCompaction(TestDataSource.WIKI, createSupervisor());
final CompactionSimulateResult simulateResult = scheduler.simulateRunWithConfigUpdate(
new ClusterCompactionConfig(null, null, null, null, null)
@@ -446,4 +474,16 @@ private void runScheduledJob()
executor.finishNextPendingTask();
}
+ private CompactionSupervisor createSupervisor()
+ {
+ return new CompactionSupervisorSpec(
+ InlineSchemaDataSourceCompactionConfig
+ .builder()
+ .forDataSource(TestDataSource.WIKI)
+ .withSkipOffsetFromLatest(Period.seconds(0))
+ .build(),
+ false,
+ scheduler
+ ).createSupervisor();
+ }
}
diff --git a/processing/src/main/java/org/apache/druid/data/output/OutputDestination.java b/processing/src/main/java/org/apache/druid/data/output/OutputDestination.java
new file mode 100644
index 000000000000..39721de67500
--- /dev/null
+++ b/processing/src/main/java/org/apache/druid/data/output/OutputDestination.java
@@ -0,0 +1,33 @@
+/*
+ * 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.data.output;
+
+import com.fasterxml.jackson.annotation.JsonSubTypes;
+import com.fasterxml.jackson.annotation.JsonTypeInfo;
+import org.apache.druid.data.input.InputSource;
+
+/**
+ * Destination where data is written out.
+ */
+@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = InputSource.TYPE_PROPERTY)
+@JsonSubTypes(value = {})
+public interface OutputDestination
+{
+}
diff --git a/processing/src/main/java/org/apache/druid/java/util/common/Intervals.java b/processing/src/main/java/org/apache/druid/java/util/common/Intervals.java
index 623f546349f9..4234a34090ee 100644
--- a/processing/src/main/java/org/apache/druid/java/util/common/Intervals.java
+++ b/processing/src/main/java/org/apache/druid/java/util/common/Intervals.java
@@ -27,6 +27,7 @@
import org.joda.time.chrono.ISOChronology;
import javax.annotation.Nullable;
+import java.util.List;
public final class Intervals
{
@@ -98,6 +99,18 @@ public static Interval findOverlappingInterval(Interval searchInterval, Interval
return null;
}
+ public static List complementOf(Interval interval)
+ {
+ if (interval.equals(Intervals.ETERNITY)) {
+ return List.of();
+ } else {
+ return List.of(
+ new Interval(DateTimes.MIN, interval.getStart()),
+ new Interval(interval.getEnd(), DateTimes.MAX)
+ );
+ }
+ }
+
private Intervals()
{
}
diff --git a/server/src/main/java/org/apache/druid/catalog/model/SchemaRegistryImpl.java b/server/src/main/java/org/apache/druid/catalog/model/SchemaRegistryImpl.java
index ff3b25b45e7b..21de49dc73c3 100644
--- a/server/src/main/java/org/apache/druid/catalog/model/SchemaRegistryImpl.java
+++ b/server/src/main/java/org/apache/druid/catalog/model/SchemaRegistryImpl.java
@@ -19,13 +19,13 @@
package org.apache.druid.catalog.model;
-import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Lists;
import org.apache.druid.catalog.model.table.DatasourceDefn;
import org.apache.druid.catalog.model.table.ExternalTableDefn;
+import org.apache.druid.catalog.model.table.IndexingTemplateDefn;
import org.apache.druid.server.security.ResourceType;
-import java.util.Collections;
+import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -96,7 +96,7 @@ public SchemaRegistryImpl()
register(new SchemaDefnImpl(
TableId.DRUID_SCHEMA,
ResourceType.DATASOURCE,
- ImmutableSet.of(DatasourceDefn.TABLE_TYPE)
+ Set.of(DatasourceDefn.TABLE_TYPE)
));
register(new SchemaDefnImpl(
TableId.LOOKUP_SCHEMA,
@@ -116,13 +116,18 @@ public SchemaRegistryImpl()
register(new SchemaDefnImpl(
TableId.EXTERNAL_SCHEMA,
EXTERNAL_RESOURCE,
- ImmutableSet.of(ExternalTableDefn.TABLE_TYPE)
+ Set.of(ExternalTableDefn.TABLE_TYPE)
));
register(new SchemaDefnImpl(
TableId.VIEW_SCHEMA,
ResourceType.VIEW,
null // TODO
));
+ register(new SchemaDefnImpl(
+ TableId.INDEXING_TEMPLATE_SCHEMA,
+ ResourceType.CONFIG,
+ Set.of(IndexingTemplateDefn.TYPE)
+ ));
}
private void register(SchemaSpec schemaDefn)
@@ -148,7 +153,7 @@ public List schemas()
// No real need to sort every time. However, this is used infrequently,
// so OK for now.
List schemas = Lists.newArrayList(builtIns.values());
- Collections.sort(schemas, (s1, s2) -> s1.name().compareTo(s2.name()));
+ schemas.sort(Comparator.comparing(SchemaSpec::name));
return schemas;
}
}
diff --git a/server/src/main/java/org/apache/druid/catalog/model/TableDefnRegistry.java b/server/src/main/java/org/apache/druid/catalog/model/TableDefnRegistry.java
index 1a14d7403937..ea442d7266cc 100644
--- a/server/src/main/java/org/apache/druid/catalog/model/TableDefnRegistry.java
+++ b/server/src/main/java/org/apache/druid/catalog/model/TableDefnRegistry.java
@@ -25,6 +25,7 @@
import org.apache.druid.catalog.model.table.DatasourceDefn;
import org.apache.druid.catalog.model.table.ExternalTableDefn;
import org.apache.druid.catalog.model.table.HttpInputSourceDefn;
+import org.apache.druid.catalog.model.table.IndexingTemplateDefn;
import org.apache.druid.catalog.model.table.InlineInputSourceDefn;
import org.apache.druid.catalog.model.table.InputFormatDefn;
import org.apache.druid.catalog.model.table.InputFormats;
@@ -63,7 +64,8 @@ public class TableDefnRegistry
// Guice later to allow extensions to define table types.
private static final List BUILTIN_TABLE_DEFNS = Arrays.asList(
new DatasourceDefn(),
- new ExternalTableDefn()
+ new ExternalTableDefn(),
+ new IndexingTemplateDefn()
);
private static final List BUILTIN_INPUT_SOURCE_DEFNS = Arrays.asList(
new InlineInputSourceDefn(),
diff --git a/server/src/main/java/org/apache/druid/catalog/model/TableId.java b/server/src/main/java/org/apache/druid/catalog/model/TableId.java
index 55fcc797561b..77265c5d7d0b 100644
--- a/server/src/main/java/org/apache/druid/catalog/model/TableId.java
+++ b/server/src/main/java/org/apache/druid/catalog/model/TableId.java
@@ -41,6 +41,11 @@ public class TableId
// Extra for views
public static final String VIEW_SCHEMA = "view";
+ /**
+ * Schema for indexing templates.
+ */
+ public static final String INDEXING_TEMPLATE_SCHEMA = "index_template";
+
private final String schema;
private final String name;
diff --git a/server/src/main/java/org/apache/druid/catalog/model/table/IndexingTemplateDefn.java b/server/src/main/java/org/apache/druid/catalog/model/table/IndexingTemplateDefn.java
new file mode 100644
index 000000000000..f11423ddb60e
--- /dev/null
+++ b/server/src/main/java/org/apache/druid/catalog/model/table/IndexingTemplateDefn.java
@@ -0,0 +1,65 @@
+/*
+ * 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.catalog.model.table;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import org.apache.druid.catalog.model.ModelProperties;
+import org.apache.druid.catalog.model.TableDefn;
+import org.apache.druid.indexing.template.BatchIndexingJobTemplate;
+
+import java.util.List;
+
+/**
+ * Definition for indexing templates.
+ */
+public class IndexingTemplateDefn extends TableDefn
+{
+ public static final String TYPE = "index_template";
+
+ /**
+ * Property to contain template payload of type {@link BatchIndexingJobTemplate}.
+ *
+ * @see PayloadProperty#TYPE_REF
+ */
+ public static final String PROPERTY_PAYLOAD = "payload";
+
+ public IndexingTemplateDefn()
+ {
+ super(
+ "Ingestion Template",
+ TYPE,
+ List.of(new PayloadProperty()),
+ null
+ );
+ }
+
+ /**
+ * Template payload property.
+ */
+ public static class PayloadProperty extends ModelProperties.TypeRefPropertyDefn
+ {
+ public static final TypeReference TYPE_REF = new TypeReference<>() {};
+
+ public PayloadProperty()
+ {
+ super(PROPERTY_PAYLOAD, "Payload of the batch indexing template", TYPE_REF);
+ }
+ }
+}
diff --git a/server/src/main/java/org/apache/druid/indexing/template/BatchIndexingJob.java b/server/src/main/java/org/apache/druid/indexing/template/BatchIndexingJob.java
new file mode 100644
index 000000000000..509226da108d
--- /dev/null
+++ b/server/src/main/java/org/apache/druid/indexing/template/BatchIndexingJob.java
@@ -0,0 +1,90 @@
+/*
+ * 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.indexing.template;
+
+import org.apache.druid.client.indexing.ClientTaskQuery;
+import org.apache.druid.error.InvalidInput;
+import org.apache.druid.query.http.ClientSqlQuery;
+
+import javax.annotation.Nullable;
+import java.util.Objects;
+
+/**
+ * A batch indexing job that can be launched by the Overlord as a task.
+ * A job may contain the {@link ClientTaskQuery} itself or an MSQ query that gets converted
+ * by the Broker to a {@code ControllerTask} and is then submitted to the Overlord.
+ */
+public class BatchIndexingJob
+{
+ private final boolean isMsq;
+ private final ClientSqlQuery msqQuery;
+ private final ClientTaskQuery task;
+
+ protected BatchIndexingJob(
+ @Nullable ClientTaskQuery task,
+ @Nullable ClientSqlQuery msqQuery
+ )
+ {
+ this.isMsq = task == null;
+ this.msqQuery = msqQuery;
+ this.task = task;
+
+ InvalidInput.conditionalException(
+ (task == null || msqQuery == null) && (task != null || msqQuery != null),
+ "Exactly one of 'task' or 'msqQuery' must be non-null"
+ );
+ }
+
+ /**
+ * @return MSQ query to be run in this job, if any.
+ * @throws NullPointerException if this not an MSQ job.
+ */
+ public ClientSqlQuery getNonNullMsqQuery()
+ {
+ return Objects.requireNonNull(msqQuery);
+ }
+
+ /**
+ * @return Task to be run in this job, if any.
+ * @throws NullPointerException if this is an MSQ job.
+ */
+ public ClientTaskQuery getNonNullTask()
+ {
+ return Objects.requireNonNull(task);
+ }
+
+ /**
+ * @return true if this is an MSQ job.
+ */
+ public boolean isMsq()
+ {
+ return isMsq;
+ }
+
+ @Override
+ public String toString()
+ {
+ return "BatchIndexingJob{" +
+ "isMsq=" + isMsq +
+ ", msqQuery=" + msqQuery +
+ ", task=" + task +
+ '}';
+ }
+}
diff --git a/server/src/main/java/org/apache/druid/indexing/template/BatchIndexingJobTemplate.java b/server/src/main/java/org/apache/druid/indexing/template/BatchIndexingJobTemplate.java
new file mode 100644
index 000000000000..d8e9251dfdf0
--- /dev/null
+++ b/server/src/main/java/org/apache/druid/indexing/template/BatchIndexingJobTemplate.java
@@ -0,0 +1,50 @@
+/*
+ * 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.indexing.template;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonTypeInfo;
+import org.apache.druid.data.input.InputSource;
+import org.apache.druid.data.output.OutputDestination;
+
+import java.util.List;
+
+/**
+ * ETL template to create a {@link BatchIndexingJob} that indexes data from an
+ * {@link InputSource} into an {@link OutputDestination}.
+ */
+@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type")
+public interface BatchIndexingJobTemplate
+{
+ /**
+ * Creates jobs with this template for the given input and output.
+ */
+ List createJobs(
+ InputSource source,
+ OutputDestination destination,
+ JobParams jobParams
+ );
+
+ /**
+ * Unique type name of this template used for JSON serialization.
+ */
+ @JsonProperty
+ String getType();
+}
diff --git a/server/src/main/java/org/apache/druid/indexing/template/JobParams.java b/server/src/main/java/org/apache/druid/indexing/template/JobParams.java
new file mode 100644
index 000000000000..775b63b04c6b
--- /dev/null
+++ b/server/src/main/java/org/apache/druid/indexing/template/JobParams.java
@@ -0,0 +1,30 @@
+/*
+ * 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.indexing.template;
+
+import org.joda.time.DateTime;
+
+/**
+ * Provides parameters required to create a {@link BatchIndexingJob}.
+ */
+public interface JobParams
+{
+ DateTime getScheduleStartTime();
+}
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 646319eec3b2..b6db09501967 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
@@ -22,7 +22,6 @@
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import org.apache.druid.client.DataSourcesSnapshot;
-import org.apache.druid.client.indexing.ClientCompactionTaskQuery;
import org.apache.druid.client.indexing.ClientCompactionTaskQueryTuningConfig;
import org.apache.druid.client.indexing.IndexingTotalWorkerCapacityInfo;
import org.apache.druid.client.indexing.TaskPayloadResponse;
@@ -86,16 +85,15 @@ public CompactionSimulateResult simulateRunWithConfig(
// Add a read-only wrapper over the actual status tracker so that we can
// account for the active tasks
- final CompactionStatusTracker simulationStatusTracker = new CompactionStatusTracker(null)
+ final CompactionStatusTracker simulationStatusTracker = new CompactionStatusTracker()
{
@Override
public CompactionStatus computeCompactionStatus(
CompactionCandidate candidate,
- DataSourceCompactionConfig config,
CompactionCandidateSearchPolicy searchPolicy
)
{
- return statusTracker.computeCompactionStatus(candidate, config, searchPolicy);
+ return statusTracker.computeCompactionStatus(candidate, searchPolicy);
}
@Override
@@ -123,12 +121,12 @@ public void onCompactionStatusComputed(
}
@Override
- public void onTaskSubmitted(ClientCompactionTaskQuery taskPayload, CompactionCandidate candidateSegments)
+ public void onTaskSubmitted(String taskId, CompactionCandidate candidateSegments)
{
// Add a row for each task in order of submission
final CompactionStatus status = candidateSegments.getCurrentStatus();
queuedIntervals.addRow(
- createRow(candidateSegments, taskPayload.getTuningConfig(), status == null ? "" : status.getReason())
+ createRow(candidateSegments, null, status == null ? "" : status.getReason())
);
}
};
@@ -174,9 +172,7 @@ private Object[] createRow(
row.add(candidate.getUmbrellaInterval());
row.add(candidate.numSegments());
row.add(candidate.getTotalBytes());
- if (tuningConfig != null) {
- row.add(CompactSegments.findMaxNumTaskSlotsUsedByOneNativeCompactionTask(tuningConfig));
- }
+ row.add(CompactionSlotManager.getMaxTaskSlotsForNativeCompactionTask(tuningConfig));
if (reason != null) {
row.add(reason);
}
diff --git a/server/src/main/java/org/apache/druid/server/compaction/CompactionSlotManager.java b/server/src/main/java/org/apache/druid/server/compaction/CompactionSlotManager.java
new file mode 100644
index 000000000000..efe1c8bdba28
--- /dev/null
+++ b/server/src/main/java/org/apache/druid/server/compaction/CompactionSlotManager.java
@@ -0,0 +1,358 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.server.compaction;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.druid.client.indexing.ClientCompactionTaskQuery;
+import org.apache.druid.client.indexing.ClientCompactionTaskQueryTuningConfig;
+import org.apache.druid.client.indexing.ClientMSQContext;
+import org.apache.druid.client.indexing.TaskPayloadResponse;
+import org.apache.druid.common.guava.FutureUtils;
+import org.apache.druid.indexer.CompactionEngine;
+import org.apache.druid.indexer.TaskStatus;
+import org.apache.druid.indexer.TaskStatusPlus;
+import org.apache.druid.indexer.partitions.DimensionRangePartitionsSpec;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.granularity.Granularity;
+import org.apache.druid.java.util.common.logger.Logger;
+import org.apache.druid.metadata.LockFilterPolicy;
+import org.apache.druid.rpc.indexing.OverlordClient;
+import org.apache.druid.server.coordinator.ClusterCompactionConfig;
+import org.apache.druid.server.coordinator.DataSourceCompactionConfig;
+import org.apache.druid.server.coordinator.duty.CoordinatorDutyUtils;
+import org.joda.time.Interval;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * Fetches running compaction tasks from the Overlord and tracks their compaction
+ * intervals and task slots.
+ */
+public class CompactionSlotManager
+{
+ /**
+ * Task type for native compaction tasks.
+ */
+ public static final String COMPACTION_TASK_TYPE = "compact";
+
+ private static final Logger log = new Logger(CompactionSlotManager.class);
+
+ private final OverlordClient overlordClient;
+ private final CompactionStatusTracker statusTracker;
+
+ private final Map> intervalsToSkipCompaction;
+
+ private int numAvailableTaskSlots;
+
+ public CompactionSlotManager(
+ OverlordClient overlordClient,
+ CompactionStatusTracker statusTracker,
+ ClusterCompactionConfig clusterCompactionConfig
+ )
+ {
+ this.overlordClient = overlordClient;
+ this.statusTracker = statusTracker;
+ this.numAvailableTaskSlots = getCompactionTaskCapacity(clusterCompactionConfig);
+ this.intervalsToSkipCompaction = new HashMap<>();
+ }
+
+ public int getNumAvailableTaskSlots()
+ {
+ return numAvailableTaskSlots;
+ }
+
+ public Map> getDatasourceIntervalsToSkipCompaction()
+ {
+ return intervalsToSkipCompaction;
+ }
+
+ public void reserveTaskSlots(int numSlotsToReserve)
+ {
+ numAvailableTaskSlots -= numSlotsToReserve;
+ }
+
+ /**
+ * Reserves task slots for the given task from the overall compaction task capacity.
+ */
+ public void reserveTaskSlots(ClientCompactionTaskQuery compactionTaskQuery)
+ {
+ // Note: The default compactionRunnerType used here should match the default runner used in CompactionTask when
+ // no runner is provided there.
+ CompactionEngine compactionRunnerType = compactionTaskQuery.getCompactionRunner() == null
+ ? CompactionEngine.NATIVE
+ : compactionTaskQuery.getCompactionRunner().getType();
+ if (compactionRunnerType == CompactionEngine.NATIVE) {
+ numAvailableTaskSlots -=
+ getMaxTaskSlotsForNativeCompactionTask(compactionTaskQuery.getTuningConfig());
+ } else {
+ numAvailableTaskSlots -=
+ getMaxTaskSlotsForMSQCompactionTask(compactionTaskQuery.getContext());
+ }
+ }
+
+ /**
+ * Retrieves currently running tasks of type {@link #COMPACTION_TASK_TYPE} from
+ * the Overlord.
+ *
+ * Also queries the Overlord for the status of all tasks that were submitted
+ * recently but are not active anymore. The statuses are then updated in the
+ * {@link CompactionStatusTracker}.
+ */
+ public List fetchRunningCompactionTasks()
+ {
+ // Fetch currently running compaction tasks
+ final List compactionTasks = CoordinatorDutyUtils.getStatusOfActiveTasks(
+ overlordClient,
+ status -> status != null && COMPACTION_TASK_TYPE.equals(status.getType())
+ );
+
+ final Set activeTaskIds
+ = compactionTasks.stream().map(TaskStatusPlus::getId).collect(Collectors.toSet());
+ trackStatusOfCompletedTasks(activeTaskIds);
+
+ final List runningCompactTasks = new ArrayList<>();
+ for (TaskStatusPlus status : compactionTasks) {
+ final TaskPayloadResponse response =
+ FutureUtils.getUnchecked(overlordClient.taskPayload(status.getId()), true);
+ if (response == null) {
+ throw new ISE("Could not find payload for active compaction task[%s]", status.getId());
+ } else if (!COMPACTION_TASK_TYPE.equals(response.getPayload().getType())) {
+ throw new ISE(
+ "Payload of active compaction task[%s] is of invalid type[%s]",
+ status.getId(), response.getPayload().getType()
+ );
+ }
+
+ runningCompactTasks.add((ClientCompactionTaskQuery) response.getPayload());
+ }
+
+ return runningCompactTasks;
+ }
+
+ /**
+ * Cancels a currently running compaction task only if the segment granularity
+ * has changed in the datasource compaction config. Otherwise, the task is
+ * retained and its intervals are skipped from the current round of compaction.
+ *
+ * @return true if the task was canceled, false otherwise.
+ */
+ public boolean cancelTaskOnlyIfGranularityChanged(
+ ClientCompactionTaskQuery compactionTaskQuery,
+ DataSourceCompactionConfig dataSourceCompactionConfig
+ )
+ {
+ if (dataSourceCompactionConfig == null
+ || dataSourceCompactionConfig.getGranularitySpec() == null
+ || compactionTaskQuery.getGranularitySpec() == null) {
+ skipTaskInterval(compactionTaskQuery);
+ reserveTaskSlots(compactionTaskQuery);
+ return false;
+ }
+
+ Granularity configuredSegmentGranularity = dataSourceCompactionConfig.getGranularitySpec()
+ .getSegmentGranularity();
+ Granularity taskSegmentGranularity = compactionTaskQuery.getGranularitySpec().getSegmentGranularity();
+ if (configuredSegmentGranularity == null || configuredSegmentGranularity.equals(taskSegmentGranularity)) {
+ skipTaskInterval(compactionTaskQuery);
+ reserveTaskSlots(compactionTaskQuery);
+ return false;
+ }
+
+ log.info(
+ "Cancelling task[%s] as task segmentGranularity[%s] differs from compaction config segmentGranularity[%s].",
+ compactionTaskQuery.getId(), taskSegmentGranularity, configuredSegmentGranularity
+ );
+ overlordClient.cancelTask(compactionTaskQuery.getId());
+ return true;
+ }
+
+ /**
+ * Retrieves the list of intervals locked by higher priority tasks for each datasource.
+ * Since compaction tasks submitted for these Intervals would have to wait anyway,
+ * we skip these Intervals until the next compaction run by adding them to
+ * {@link #intervalsToSkipCompaction}.
+ *
+ */
+ public void skipLockedIntervals(List compactionConfigs)
+ {
+ final List lockFilterPolicies = compactionConfigs
+ .stream()
+ .map(config ->
+ new LockFilterPolicy(config.getDataSource(), config.getTaskPriority(), null, config.getTaskContext()))
+ .collect(Collectors.toList());
+ final Map> datasourceToLockedIntervals =
+ new HashMap<>(FutureUtils.getUnchecked(overlordClient.findLockedIntervals(lockFilterPolicies), true));
+ log.debug(
+ "Skipping the following intervals for Compaction as they are currently locked: %s",
+ datasourceToLockedIntervals
+ );
+
+ // Skip all the intervals locked by higher priority tasks for each datasource
+ // This must be done after the invalid compaction tasks are cancelled
+ // in the loop above so that their intervals are not considered locked
+ datasourceToLockedIntervals.forEach(
+ (dataSource, intervals) ->
+ intervalsToSkipCompaction
+ .computeIfAbsent(dataSource, ds -> new ArrayList<>())
+ .addAll(intervals)
+ );
+ }
+
+ /**
+ * Adds the compaction interval of this task to {@link #intervalsToSkipCompaction}
+ */
+ private void skipTaskInterval(ClientCompactionTaskQuery compactionTask)
+ {
+ final Interval interval = compactionTask.getIoConfig().getInputSpec().getInterval();
+ intervalsToSkipCompaction.computeIfAbsent(compactionTask.getDataSource(), k -> new ArrayList<>())
+ .add(interval);
+ }
+
+ /**
+ * Computes overall compaction task capacity for the cluster.
+ *
+ * @return A value >= 1.
+ */
+ private int getCompactionTaskCapacity(ClusterCompactionConfig clusterConfig)
+ {
+ int totalWorkerCapacity = CoordinatorDutyUtils.getTotalWorkerCapacity(overlordClient);
+
+ int compactionTaskCapacity = Math.min(
+ (int) (totalWorkerCapacity * clusterConfig.getCompactionTaskSlotRatio()),
+ clusterConfig.getMaxCompactionTaskSlots()
+ );
+
+ // Always consider atleast one slot available for compaction
+ return Math.max(1, compactionTaskCapacity);
+ }
+
+ /**
+ * Queries the Overlord for the status of all tasks that were submitted
+ * recently but are not active anymore. The statuses are then updated in the
+ * {@link #statusTracker}.
+ */
+ private void trackStatusOfCompletedTasks(Set activeTaskIds)
+ {
+ final Set finishedTaskIds = new HashSet<>(statusTracker.getSubmittedTaskIds());
+ finishedTaskIds.removeAll(activeTaskIds);
+
+ if (finishedTaskIds.isEmpty()) {
+ return;
+ }
+
+ final Map taskStatusMap
+ = FutureUtils.getUnchecked(overlordClient.taskStatuses(finishedTaskIds), true);
+ for (String taskId : finishedTaskIds) {
+ // Assume unknown task to have finished successfully
+ final TaskStatus taskStatus = taskStatusMap.getOrDefault(taskId, TaskStatus.success(taskId));
+ if (taskStatus.isComplete()) {
+ statusTracker.onTaskFinished(taskId, taskStatus);
+ }
+ }
+ }
+
+ /**
+ * @return Maximum number of task slots used by a native compaction task at
+ * any time when the task is run with the given tuning config.
+ */
+ public static int getMaxTaskSlotsForNativeCompactionTask(
+ @Nullable ClientCompactionTaskQueryTuningConfig tuningConfig
+ )
+ {
+ if (isParallelMode(tuningConfig)) {
+ Integer maxNumConcurrentSubTasks = tuningConfig.getMaxNumConcurrentSubTasks();
+ // Max number of task slots used in parallel mode = maxNumConcurrentSubTasks + 1 (supervisor task)
+ return (maxNumConcurrentSubTasks == null ? 1 : maxNumConcurrentSubTasks) + 1;
+ } else {
+ return 1;
+ }
+ }
+
+ /**
+ * @return Maximum number of task slots used by an MSQ compaction task at any
+ * time when the task is run with the given context.
+ */
+ public static int getMaxTaskSlotsForMSQCompactionTask(@Nullable Map context)
+ {
+ return context == null
+ ? ClientMSQContext.DEFAULT_MAX_NUM_TASKS
+ : (int) context.getOrDefault(ClientMSQContext.CTX_MAX_NUM_TASKS, ClientMSQContext.DEFAULT_MAX_NUM_TASKS);
+ }
+
+
+ /**
+ * Returns true if the compaction task can run in the parallel mode with the given tuningConfig.
+ * This method should be synchronized with ParallelIndexSupervisorTask.isParallelMode(InputSource, ParallelIndexTuningConfig).
+ */
+ @VisibleForTesting
+ public static boolean isParallelMode(@Nullable ClientCompactionTaskQueryTuningConfig tuningConfig)
+ {
+ if (null == tuningConfig) {
+ return false;
+ }
+ boolean useRangePartitions = useRangePartitions(tuningConfig);
+ int minRequiredNumConcurrentSubTasks = useRangePartitions ? 1 : 2;
+ return tuningConfig.getMaxNumConcurrentSubTasks() != null
+ && tuningConfig.getMaxNumConcurrentSubTasks() >= minRequiredNumConcurrentSubTasks;
+ }
+
+ private static boolean useRangePartitions(ClientCompactionTaskQueryTuningConfig tuningConfig)
+ {
+ // dynamic partitionsSpec will be used if getPartitionsSpec() returns null
+ return tuningConfig.getPartitionsSpec() instanceof DimensionRangePartitionsSpec;
+ }
+
+ public int computeSlotsRequiredForTask(
+ ClientCompactionTaskQuery task,
+ DataSourceCompactionConfig config
+ )
+ {
+ if (task.getCompactionRunner().getType() == CompactionEngine.MSQ) {
+ final Map autoCompactionContext = task.getContext();
+ if (autoCompactionContext.containsKey(ClientMSQContext.CTX_MAX_NUM_TASKS)) {
+ return (int) autoCompactionContext.get(ClientMSQContext.CTX_MAX_NUM_TASKS);
+ } else {
+ // Since MSQ needs all task slots for the calculated #tasks to be available upfront, allot all available
+ // compaction slots (upto a max of MAX_TASK_SLOTS_FOR_MSQ_COMPACTION) to current compaction task to avoid
+ // stalling. Setting "taskAssignment" to "auto" has the problem of not being able to determine the actual
+ // count, which is required for subsequent tasks.
+ final int slotsRequiredForCurrentTask = Math.min(
+ // Update the slots to 2 (min required for MSQ) if only 1 slot is available.
+ numAvailableTaskSlots == 1 ? 2 : numAvailableTaskSlots,
+ ClientMSQContext.MAX_TASK_SLOTS_FOR_MSQ_COMPACTION_TASK
+ );
+ autoCompactionContext.put(ClientMSQContext.CTX_MAX_NUM_TASKS, slotsRequiredForCurrentTask);
+
+ return slotsRequiredForCurrentTask;
+ }
+ } else {
+ return CompactionSlotManager.getMaxTaskSlotsForNativeCompactionTask(
+ config.getTuningConfig()
+ );
+ }
+ }
+}
diff --git a/server/src/main/java/org/apache/druid/server/compaction/CompactionSnapshotBuilder.java b/server/src/main/java/org/apache/druid/server/compaction/CompactionSnapshotBuilder.java
new file mode 100644
index 000000000000..e52d24edbf01
--- /dev/null
+++ b/server/src/main/java/org/apache/druid/server/compaction/CompactionSnapshotBuilder.java
@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.server.compaction;
+
+import org.apache.druid.server.coordinator.AutoCompactionSnapshot;
+import org.apache.druid.server.coordinator.stats.CoordinatorRunStats;
+import org.apache.druid.server.coordinator.stats.Dimension;
+import org.apache.druid.server.coordinator.stats.RowKey;
+import org.apache.druid.server.coordinator.stats.Stats;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Builds {@link AutoCompactionSnapshot} for multiple datasources using the
+ * identified {@link CompactionCandidate} list.
+ */
+public class CompactionSnapshotBuilder
+{
+ private final CoordinatorRunStats stats;
+ private final Map datasourceToBuilder = new HashMap<>();
+
+ public CompactionSnapshotBuilder(CoordinatorRunStats runStats)
+ {
+ this.stats = runStats;
+ }
+
+ public void addToComplete(CompactionCandidate candidate)
+ {
+ getBuilderForDatasource(candidate.getDataSource())
+ .incrementCompactedStats(candidate.getStats());
+ }
+
+ public void addToPending(CompactionCandidate candidate)
+ {
+ getBuilderForDatasource(candidate.getDataSource())
+ .incrementWaitingStats(candidate.getStats());
+ }
+
+ public void addToSkipped(CompactionCandidate candidate)
+ {
+ getBuilderForDatasource(candidate.getDataSource())
+ .incrementSkippedStats(candidate.getStats());
+ }
+
+ public Map build()
+ {
+ final Map datasourceToSnapshot = new HashMap<>();
+ datasourceToBuilder.forEach((dataSource, builder) -> {
+ final AutoCompactionSnapshot autoCompactionSnapshot = builder.build();
+ datasourceToSnapshot.put(dataSource, autoCompactionSnapshot);
+ collectSnapshotStats(autoCompactionSnapshot);
+ });
+
+ return datasourceToSnapshot;
+ }
+
+ private AutoCompactionSnapshot.Builder getBuilderForDatasource(String dataSource)
+ {
+ return datasourceToBuilder.computeIfAbsent(dataSource, AutoCompactionSnapshot::builder);
+ }
+
+ private void collectSnapshotStats(AutoCompactionSnapshot autoCompactionSnapshot)
+ {
+ final RowKey rowKey = RowKey.of(Dimension.DATASOURCE, autoCompactionSnapshot.getDataSource());
+
+ stats.add(Stats.Compaction.PENDING_BYTES, rowKey, autoCompactionSnapshot.getBytesAwaitingCompaction());
+ stats.add(Stats.Compaction.PENDING_SEGMENTS, rowKey, autoCompactionSnapshot.getSegmentCountAwaitingCompaction());
+ stats.add(Stats.Compaction.PENDING_INTERVALS, rowKey, autoCompactionSnapshot.getIntervalCountAwaitingCompaction());
+ stats.add(Stats.Compaction.COMPACTED_BYTES, rowKey, autoCompactionSnapshot.getBytesCompacted());
+ stats.add(Stats.Compaction.COMPACTED_SEGMENTS, rowKey, autoCompactionSnapshot.getSegmentCountCompacted());
+ stats.add(Stats.Compaction.COMPACTED_INTERVALS, rowKey, autoCompactionSnapshot.getIntervalCountCompacted());
+ stats.add(Stats.Compaction.SKIPPED_BYTES, rowKey, autoCompactionSnapshot.getBytesSkipped());
+ stats.add(Stats.Compaction.SKIPPED_SEGMENTS, rowKey, autoCompactionSnapshot.getSegmentCountSkipped());
+ stats.add(Stats.Compaction.SKIPPED_INTERVALS, rowKey, autoCompactionSnapshot.getIntervalCountSkipped());
+ }
+}
diff --git a/server/src/main/java/org/apache/druid/server/compaction/CompactionStatus.java b/server/src/main/java/org/apache/druid/server/compaction/CompactionStatus.java
index 98194eefdbd5..97b645388218 100644
--- a/server/src/main/java/org/apache/druid/server/compaction/CompactionStatus.java
+++ b/server/src/main/java/org/apache/druid/server/compaction/CompactionStatus.java
@@ -19,12 +19,12 @@
package org.apache.druid.server.compaction;
-import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.druid.client.indexing.ClientCompactionTaskGranularitySpec;
import org.apache.druid.client.indexing.ClientCompactionTaskQueryTuningConfig;
import org.apache.druid.common.config.Configs;
import org.apache.druid.data.input.impl.DimensionSchema;
+import org.apache.druid.indexer.granularity.GranularitySpec;
import org.apache.druid.indexer.partitions.DimensionRangePartitionsSpec;
import org.apache.druid.indexer.partitions.DynamicPartitionsSpec;
import org.apache.druid.indexer.partitions.HashedPartitionsSpec;
@@ -64,6 +64,7 @@ public enum State
* The order of the checks must be honored while evaluating them.
*/
private static final List> CHECKS = Arrays.asList(
+ Evaluator::inputBytesAreWithinLimit,
Evaluator::segmentsHaveBeenCompactedAtLeastOnce,
Evaluator::allCandidatesHaveSameCompactionState,
Evaluator::partitionsSpecIsUpToDate,
@@ -115,7 +116,7 @@ public String toString()
'}';
}
- private static CompactionStatus incomplete(String reasonFormat, Object... args)
+ public static CompactionStatus pending(String reasonFormat, Object... args)
{
return new CompactionStatus(State.PENDING, StringUtils.format(reasonFormat, args));
}
@@ -141,7 +142,7 @@ private static CompactionStatus configChanged(
Function stringFunction
)
{
- return CompactionStatus.incomplete(
+ return CompactionStatus.pending(
"'%s' mismatch: required[%s], current[%s]",
field,
target == null ? null : stringFunction.apply(target),
@@ -204,11 +205,10 @@ static CompactionStatus running(String reasonForCompaction)
*/
static CompactionStatus compute(
CompactionCandidate candidateSegments,
- DataSourceCompactionConfig config,
- ObjectMapper objectMapper
+ DataSourceCompactionConfig config
)
{
- final Evaluator evaluator = new Evaluator(candidateSegments, config, objectMapper);
+ final Evaluator evaluator = new Evaluator(candidateSegments, config);
return CHECKS.stream().map(f -> f.apply(evaluator))
.filter(status -> !status.isComplete())
.findFirst().orElse(COMPLETE);
@@ -266,7 +266,6 @@ static DimensionRangePartitionsSpec getEffectiveRangePartitionsSpec(DimensionRan
*/
private static class Evaluator
{
- private final ObjectMapper objectMapper;
private final DataSourceCompactionConfig compactionConfig;
private final CompactionCandidate candidateSegments;
private final CompactionState lastCompactionState;
@@ -276,12 +275,10 @@ private static class Evaluator
private Evaluator(
CompactionCandidate candidateSegments,
- DataSourceCompactionConfig compactionConfig,
- ObjectMapper objectMapper
+ DataSourceCompactionConfig compactionConfig
)
{
this.candidateSegments = candidateSegments;
- this.objectMapper = objectMapper;
this.lastCompactionState = candidateSegments.getSegments().get(0).getLastCompactionState();
this.compactionConfig = compactionConfig;
this.tuningConfig = ClientCompactionTaskQueryTuningConfig.from(compactionConfig);
@@ -290,8 +287,7 @@ private Evaluator(
this.existingGranularitySpec = null;
} else {
this.existingGranularitySpec = convertIfNotNull(
- lastCompactionState.getGranularitySpec(),
- ClientCompactionTaskGranularitySpec.class
+ lastCompactionState.getGranularitySpec()
);
}
}
@@ -299,7 +295,7 @@ private Evaluator(
private CompactionStatus segmentsHaveBeenCompactedAtLeastOnce()
{
if (lastCompactionState == null) {
- return CompactionStatus.incomplete("not compacted yet");
+ return CompactionStatus.pending("not compacted yet");
} else {
return COMPLETE;
}
@@ -313,7 +309,7 @@ private CompactionStatus allCandidatesHaveSameCompactionState()
if (allHaveSameCompactionState) {
return COMPLETE;
} else {
- return CompactionStatus.incomplete("segments have different last compaction states");
+ return CompactionStatus.pending("segments have different last compaction states");
}
}
@@ -336,7 +332,7 @@ private CompactionStatus indexSpecIsUpToDate()
return CompactionStatus.completeIfEqual(
"indexSpec",
Configs.valueOrDefault(tuningConfig.getIndexSpec(), IndexSpec.DEFAULT),
- objectMapper.convertValue(lastCompactionState.getIndexSpec(), IndexSpec.class),
+ lastCompactionState.getIndexSpec(),
String::valueOf
);
}
@@ -351,6 +347,19 @@ private CompactionStatus projectionsAreUpToDate()
);
}
+ private CompactionStatus inputBytesAreWithinLimit()
+ {
+ final long inputSegmentSize = compactionConfig.getInputSegmentSizeBytes();
+ if (candidateSegments.getTotalBytes() > inputSegmentSize) {
+ return CompactionStatus.skipped(
+ "'inputSegmentSize' exceeded: Total segment size[%d] is larger than allowed inputSegmentSize[%d]",
+ candidateSegments.getTotalBytes(), inputSegmentSize
+ );
+ } else {
+ return COMPLETE;
+ }
+ }
+
private CompactionStatus segmentGranularityIsUpToDate()
{
if (configuredGranularitySpec == null
@@ -371,7 +380,7 @@ private CompactionStatus segmentGranularityIsUpToDate()
segment -> !configuredSegmentGranularity.isAligned(segment.getInterval())
);
if (needsCompaction) {
- return CompactionStatus.incomplete(
+ return CompactionStatus.pending(
"segmentGranularity: segments do not align with target[%s]",
asString(configuredSegmentGranularity)
);
@@ -477,10 +486,7 @@ private CompactionStatus transformSpecFilterIsUpToDate()
return COMPLETE;
}
- CompactionTransformSpec existingTransformSpec = convertIfNotNull(
- lastCompactionState.getTransformSpec(),
- CompactionTransformSpec.class
- );
+ CompactionTransformSpec existingTransformSpec = lastCompactionState.getTransformSpec();
return CompactionStatus.completeIfEqual(
"transformSpec filter",
compactionConfig.getTransformSpec().getFilter(),
@@ -490,12 +496,16 @@ private CompactionStatus transformSpecFilterIsUpToDate()
}
@Nullable
- private T convertIfNotNull(Object object, Class clazz)
+ private ClientCompactionTaskGranularitySpec convertIfNotNull(GranularitySpec granularitySpec)
{
- if (object == null) {
+ if (granularitySpec == null) {
return null;
} else {
- return objectMapper.convertValue(object, clazz);
+ return new ClientCompactionTaskGranularitySpec(
+ granularitySpec.getSegmentGranularity(),
+ granularitySpec.getQueryGranularity(),
+ granularitySpec.isRollup()
+ );
}
}
}
diff --git a/server/src/main/java/org/apache/druid/server/compaction/CompactionStatusTracker.java b/server/src/main/java/org/apache/druid/server/compaction/CompactionStatusTracker.java
index cbf5f25f9d7b..5c3daf2a26d9 100644
--- a/server/src/main/java/org/apache/druid/server/compaction/CompactionStatusTracker.java
+++ b/server/src/main/java/org/apache/druid/server/compaction/CompactionStatusTracker.java
@@ -19,14 +19,10 @@
package org.apache.druid.server.compaction;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.google.inject.Inject;
-import org.apache.druid.client.indexing.ClientCompactionTaskQuery;
import org.apache.druid.indexer.TaskState;
import org.apache.druid.indexer.TaskStatus;
import org.apache.druid.java.util.common.DateTimes;
import org.apache.druid.server.coordinator.DataSourceCompactionConfig;
-import org.apache.druid.server.coordinator.DruidCompactionConfig;
import org.joda.time.DateTime;
import org.joda.time.Duration;
import org.joda.time.Interval;
@@ -45,7 +41,6 @@ public class CompactionStatusTracker
{
private static final Duration MAX_STATUS_RETAIN_DURATION = Duration.standardHours(12);
- private final ObjectMapper objectMapper;
private final ConcurrentHashMap datasourceStatuses
= new ConcurrentHashMap<>();
private final ConcurrentHashMap submittedTaskIdToSegments
@@ -53,10 +48,8 @@ public class CompactionStatusTracker
private final AtomicReference segmentSnapshotTime = new AtomicReference<>();
- @Inject
- public CompactionStatusTracker(ObjectMapper objectMapper)
+ public CompactionStatusTracker()
{
- this.objectMapper = objectMapper;
}
public void stop()
@@ -86,25 +79,17 @@ public Set getSubmittedTaskIds()
return submittedTaskIdToSegments.keySet();
}
+ /**
+ * Checks if compaction can be started for the given {@link CompactionCandidate}.
+ * This method assumes that the given candidate is eligible for compaction
+ * based on the current compaction config/supervisor of the datasource.
+ */
public CompactionStatus computeCompactionStatus(
CompactionCandidate candidate,
- DataSourceCompactionConfig config,
CompactionCandidateSearchPolicy searchPolicy
)
{
- final CompactionStatus compactionStatus = CompactionStatus.compute(candidate, config, objectMapper);
- if (compactionStatus.isComplete()) {
- return compactionStatus;
- }
-
- // Skip intervals that violate max allowed input segment size
- final long inputSegmentSize = config.getInputSegmentSizeBytes();
- if (candidate.getTotalBytes() > inputSegmentSize) {
- return CompactionStatus.skipped(
- "'inputSegmentSize' exceeded: Total segment size[%d] is larger than allowed inputSegmentSize[%d]",
- candidate.getTotalBytes(), inputSegmentSize
- );
- }
+ final CompactionStatus pendingStatus = CompactionStatus.pending("not compacted yet");
// Skip intervals that already have a running task
final CompactionTaskStatus lastTaskStatus = getLatestTaskStatus(candidate);
@@ -123,11 +108,11 @@ public CompactionStatus computeCompactionStatus(
}
// Skip intervals that have been filtered out by the policy
- if (!searchPolicy.isEligibleForCompaction(candidate, compactionStatus, lastTaskStatus)) {
+ if (!searchPolicy.isEligibleForCompaction(candidate, pendingStatus, lastTaskStatus)) {
return CompactionStatus.skipped("Rejected by search policy");
}
- return compactionStatus;
+ return pendingStatus;
}
public void onCompactionStatusComputed(
@@ -143,17 +128,15 @@ public void onSegmentTimelineUpdated(DateTime snapshotTime)
this.segmentSnapshotTime.set(snapshotTime);
}
- public void onCompactionConfigUpdated(DruidCompactionConfig compactionConfig)
+ /**
+ * Updates the set of datasources that have compaction enabled and cleans up
+ * stale task statuses.
+ */
+ public void resetActiveDatasources(Set compactionEnabledDatasources)
{
- final Set compactionEnabledDatasources = new HashSet<>();
- if (compactionConfig.getCompactionConfigs() != null) {
- compactionConfig.getCompactionConfigs().forEach(config -> {
- getOrComputeDatasourceStatus(config.getDataSource())
- .cleanupStaleTaskStatuses();
-
- compactionEnabledDatasources.add(config.getDataSource());
- });
- }
+ compactionEnabledDatasources.forEach(
+ dataSource -> getOrComputeDatasourceStatus(dataSource).cleanupStaleTaskStatuses()
+ );
// Clean up state for datasources where compaction has been disabled
final Set allDatasources = new HashSet<>(datasourceStatuses.keySet());
@@ -165,12 +148,12 @@ public void onCompactionConfigUpdated(DruidCompactionConfig compactionConfig)
}
public void onTaskSubmitted(
- ClientCompactionTaskQuery taskPayload,
+ String taskId,
CompactionCandidate candidateSegments
)
{
- submittedTaskIdToSegments.put(taskPayload.getId(), candidateSegments);
- getOrComputeDatasourceStatus(taskPayload.getDataSource())
+ submittedTaskIdToSegments.put(taskId, candidateSegments);
+ getOrComputeDatasourceStatus(candidateSegments.getDataSource())
.handleSubmittedTask(candidateSegments);
}
diff --git a/server/src/main/java/org/apache/druid/server/compaction/DataSourceCompactibleSegmentIterator.java b/server/src/main/java/org/apache/druid/server/compaction/DataSourceCompactibleSegmentIterator.java
index e15d310b33db..588e4ba77d41 100644
--- a/server/src/main/java/org/apache/druid/server/compaction/DataSourceCompactibleSegmentIterator.java
+++ b/server/src/main/java/org/apache/druid/server/compaction/DataSourceCompactibleSegmentIterator.java
@@ -65,8 +65,6 @@ public class DataSourceCompactibleSegmentIterator implements CompactionSegmentIt
private final String dataSource;
private final DataSourceCompactionConfig config;
- private final CompactionStatusTracker statusTracker;
- private final CompactionCandidateSearchPolicy searchPolicy;
private final List compactedSegments = new ArrayList<>();
private final List skippedSegments = new ArrayList<>();
@@ -83,14 +81,11 @@ public DataSourceCompactibleSegmentIterator(
DataSourceCompactionConfig config,
SegmentTimeline timeline,
List skipIntervals,
- CompactionCandidateSearchPolicy searchPolicy,
- CompactionStatusTracker statusTracker
+ CompactionCandidateSearchPolicy searchPolicy
)
{
- this.statusTracker = statusTracker;
this.config = config;
this.dataSource = config.getDataSource();
- this.searchPolicy = searchPolicy;
this.queue = new PriorityQueue<>(searchPolicy::compareCandidates);
populateQueue(timeline, skipIntervals);
@@ -121,7 +116,6 @@ private void populateQueue(SegmentTimeline timeline, List skipInterval
CompactionStatus.skipped("Segments have partial-eternity intervals")
);
skippedSegments.add(candidatesWithStatus);
- statusTracker.onCompactionStatusComputed(candidatesWithStatus, config);
return;
}
@@ -316,10 +310,8 @@ private void findAndEnqueueSegmentsToCompact(CompactibleSegmentIterator compacti
}
final CompactionCandidate candidates = CompactionCandidate.from(segments);
- final CompactionStatus compactionStatus
- = statusTracker.computeCompactionStatus(candidates, config, searchPolicy);
+ final CompactionStatus compactionStatus = CompactionStatus.compute(candidates, config);
final CompactionCandidate candidatesWithStatus = candidates.withCurrentStatus(compactionStatus);
- statusTracker.onCompactionStatusComputed(candidatesWithStatus, config);
if (compactionStatus.isComplete()) {
compactedSegments.add(candidatesWithStatus);
@@ -371,7 +363,6 @@ private List findInitialSearchInterval(
final CompactionCandidate candidatesWithStatus = candidates.withCurrentStatus(reason);
skippedSegments.add(candidatesWithStatus);
- statusTracker.onCompactionStatusComputed(candidatesWithStatus, config);
}
}
diff --git a/server/src/main/java/org/apache/druid/server/compaction/PriorityBasedCompactionSegmentIterator.java b/server/src/main/java/org/apache/druid/server/compaction/PriorityBasedCompactionSegmentIterator.java
index 1e91df7e38a0..49d936fda0ac 100644
--- a/server/src/main/java/org/apache/druid/server/compaction/PriorityBasedCompactionSegmentIterator.java
+++ b/server/src/main/java/org/apache/druid/server/compaction/PriorityBasedCompactionSegmentIterator.java
@@ -48,8 +48,7 @@ public PriorityBasedCompactionSegmentIterator(
CompactionCandidateSearchPolicy searchPolicy,
Map compactionConfigs,
Map datasourceToTimeline,
- Map> skipIntervals,
- CompactionStatusTracker statusTracker
+ Map> skipIntervals
)
{
this.queue = new PriorityQueue<>(searchPolicy::compareCandidates);
@@ -70,8 +69,7 @@ public PriorityBasedCompactionSegmentIterator(
compactionConfigs.get(datasource),
timeline,
skipIntervals.getOrDefault(datasource, Collections.emptyList()),
- searchPolicy,
- statusTracker
+ searchPolicy
)
);
addNextItemForDatasourceToQueue(datasource);
diff --git a/server/src/main/java/org/apache/druid/server/coordinator/duty/CompactSegments.java b/server/src/main/java/org/apache/druid/server/coordinator/duty/CompactSegments.java
index 543bc2b82139..cf166c8f3204 100644
--- a/server/src/main/java/org/apache/druid/server/coordinator/duty/CompactSegments.java
+++ b/server/src/main/java/org/apache/druid/server/coordinator/duty/CompactSegments.java
@@ -23,7 +23,6 @@
import com.fasterxml.jackson.annotation.JsonCreator;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
-import com.google.common.base.Predicate;
import org.apache.druid.client.DataSourcesSnapshot;
import org.apache.druid.client.indexing.ClientCompactionIOConfig;
import org.apache.druid.client.indexing.ClientCompactionIntervalSpec;
@@ -32,26 +31,22 @@
import org.apache.druid.client.indexing.ClientCompactionTaskGranularitySpec;
import org.apache.druid.client.indexing.ClientCompactionTaskQuery;
import org.apache.druid.client.indexing.ClientCompactionTaskQueryTuningConfig;
-import org.apache.druid.client.indexing.ClientMSQContext;
-import org.apache.druid.client.indexing.TaskPayloadResponse;
import org.apache.druid.common.guava.FutureUtils;
import org.apache.druid.common.utils.IdUtils;
import org.apache.druid.data.input.impl.AggregateProjectionSpec;
import org.apache.druid.indexer.CompactionEngine;
-import org.apache.druid.indexer.TaskStatus;
-import org.apache.druid.indexer.TaskStatusPlus;
-import org.apache.druid.indexer.partitions.DimensionRangePartitionsSpec;
-import org.apache.druid.java.util.common.ISE;
import org.apache.druid.java.util.common.granularity.Granularity;
import org.apache.druid.java.util.common.granularity.GranularityType;
import org.apache.druid.java.util.common.logger.Logger;
-import org.apache.druid.metadata.LockFilterPolicy;
import org.apache.druid.query.aggregation.AggregatorFactory;
import org.apache.druid.rpc.indexing.OverlordClient;
import org.apache.druid.segment.transform.CompactionTransformSpec;
import org.apache.druid.server.compaction.CompactionCandidate;
import org.apache.druid.server.compaction.CompactionCandidateSearchPolicy;
import org.apache.druid.server.compaction.CompactionSegmentIterator;
+import org.apache.druid.server.compaction.CompactionSlotManager;
+import org.apache.druid.server.compaction.CompactionSnapshotBuilder;
+import org.apache.druid.server.compaction.CompactionStatus;
import org.apache.druid.server.compaction.CompactionStatusTracker;
import org.apache.druid.server.compaction.PriorityBasedCompactionSegmentIterator;
import org.apache.druid.server.coordinator.AutoCompactionSnapshot;
@@ -66,10 +61,8 @@
import org.joda.time.Interval;
import javax.annotation.Nullable;
-import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
-import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -79,10 +72,6 @@
public class CompactSegments implements CoordinatorCustomDuty
{
- /**
- * Must be the same as org.apache.druid.indexing.common.task.CompactionTask.TYPE.
- */
- public static final String COMPACTION_TASK_TYPE = "compact";
/**
* Must be the same as org.apache.druid.indexing.common.task.Tasks.STORE_COMPACTION_STATE_KEY
*/
@@ -92,8 +81,6 @@ public class CompactSegments implements CoordinatorCustomDuty
private static final Logger LOG = new Logger(CompactSegments.class);
private static final String TASK_ID_PREFIX = "coordinator-issued";
- private static final Predicate IS_COMPACTION_TASK =
- status -> null != status && COMPACTION_TASK_TYPE.equals(status.getType());
private final CompactionStatusTracker statusTracker;
private final OverlordClient overlordClient;
@@ -152,75 +139,36 @@ public void run(
}
statusTracker.onSegmentTimelineUpdated(dataSources.getSnapshotTime());
- statusTracker.onCompactionConfigUpdated(dynamicConfig);
List compactionConfigList = dynamicConfig.getCompactionConfigs();
if (compactionConfigList == null || compactionConfigList.isEmpty()) {
resetCompactionSnapshot();
+ statusTracker.resetActiveDatasources(Set.of());
return;
}
Map compactionConfigs = compactionConfigList
.stream()
.collect(Collectors.toMap(DataSourceCompactionConfig::getDataSource, Function.identity()));
+ statusTracker.resetActiveDatasources(compactionConfigs.keySet());
- // Map from dataSource to list of intervals for which compaction will be skipped in this run
- final Map> intervalsToSkipCompaction = new HashMap<>();
-
- // Fetch currently running compaction tasks
- int busyCompactionTaskSlots = 0;
- final List compactionTasks = CoordinatorDutyUtils.getStatusOfActiveTasks(
+ final CompactionSlotManager slotManager = new CompactionSlotManager(
overlordClient,
- IS_COMPACTION_TASK
+ statusTracker,
+ dynamicConfig.clusterConfig()
);
+ stats.add(Stats.Compaction.MAX_SLOTS, slotManager.getNumAvailableTaskSlots());
- final Set activeTaskIds
- = compactionTasks.stream().map(TaskStatusPlus::getId).collect(Collectors.toSet());
- trackStatusOfCompletedTasks(activeTaskIds);
-
- for (TaskStatusPlus status : compactionTasks) {
- final TaskPayloadResponse response =
- FutureUtils.getUnchecked(overlordClient.taskPayload(status.getId()), true);
- if (response == null) {
- throw new ISE("Could not find payload for active compaction task[%s]", status.getId());
- } else if (!COMPACTION_TASK_TYPE.equals(response.getPayload().getType())) {
- throw new ISE(
- "Payload of active compaction task[%s] is of invalid type[%s]",
- status.getId(), response.getPayload().getType()
- );
- }
-
- final ClientCompactionTaskQuery compactionTaskQuery = (ClientCompactionTaskQuery) response.getPayload();
- DataSourceCompactionConfig dataSourceCompactionConfig = compactionConfigs.get(status.getDataSource());
- if (cancelTaskIfGranularityChanged(compactionTaskQuery, dataSourceCompactionConfig)) {
- continue;
- }
-
- // Skip this interval as the current active compaction task is good
- final Interval interval = compactionTaskQuery.getIoConfig().getInputSpec().getInterval();
- intervalsToSkipCompaction.computeIfAbsent(status.getDataSource(), k -> new ArrayList<>())
- .add(interval);
- // Note: The default compactionRunnerType used here should match the default runner used in CompactionTask when
- // no runner is provided there.
- CompactionEngine compactionRunnerType = compactionTaskQuery.getCompactionRunner() == null
- ? CompactionEngine.NATIVE
- : compactionTaskQuery.getCompactionRunner().getType();
- if (compactionRunnerType == CompactionEngine.NATIVE) {
- busyCompactionTaskSlots +=
- findMaxNumTaskSlotsUsedByOneNativeCompactionTask(compactionTaskQuery.getTuningConfig());
- } else {
- busyCompactionTaskSlots += findMaxNumTaskSlotsUsedByOneMsqCompactionTask(compactionTaskQuery.getContext());
+ // Fetch currently running compaction tasks
+ for (ClientCompactionTaskQuery compactionTaskQuery : slotManager.fetchRunningCompactionTasks()) {
+ final String dataSource = compactionTaskQuery.getDataSource();
+ DataSourceCompactionConfig dataSourceCompactionConfig = compactionConfigs.get(dataSource);
+ if (slotManager.cancelTaskOnlyIfGranularityChanged(compactionTaskQuery, dataSourceCompactionConfig)) {
+ stats.add(Stats.Compaction.CANCELLED_TASKS, RowKey.of(Dimension.DATASOURCE, dataSource), 1L);
}
}
+ stats.add(Stats.Compaction.AVAILABLE_SLOTS, slotManager.getNumAvailableTaskSlots());
- // Skip all the intervals locked by higher priority tasks for each datasource
- // This must be done after the invalid compaction tasks are cancelled
- // in the loop above so that their intervals are not considered locked
- getLockedIntervals(compactionConfigList).forEach(
- (dataSource, intervals) ->
- intervalsToSkipCompaction
- .computeIfAbsent(dataSource, ds -> new ArrayList<>())
- .addAll(intervals)
- );
+ slotManager.skipLockedIntervals(compactionConfigList);
// Get iterator over segments to compact and submit compaction tasks
final CompactionCandidateSearchPolicy policy = dynamicConfig.getCompactionPolicy();
@@ -228,27 +176,21 @@ public void run(
policy,
compactionConfigs,
dataSources.getUsedSegmentsTimelinesPerDataSource(),
- intervalsToSkipCompaction,
- statusTracker
+ slotManager.getDatasourceIntervalsToSkipCompaction()
);
- final int compactionTaskCapacity = getCompactionTaskCapacity(dynamicConfig);
- final int availableCompactionTaskSlots
- = getAvailableCompactionTaskSlots(compactionTaskCapacity, busyCompactionTaskSlots);
-
- final Map currentRunAutoCompactionSnapshotBuilders = new HashMap<>();
+ final CompactionSnapshotBuilder compactionSnapshotBuilder = new CompactionSnapshotBuilder(stats);
final int numSubmittedCompactionTasks = submitCompactionTasks(
compactionConfigs,
- currentRunAutoCompactionSnapshotBuilders,
- availableCompactionTaskSlots,
+ compactionSnapshotBuilder,
+ slotManager,
iterator,
+ policy,
defaultEngine
);
- stats.add(Stats.Compaction.MAX_SLOTS, compactionTaskCapacity);
- stats.add(Stats.Compaction.AVAILABLE_SLOTS, availableCompactionTaskSlots);
stats.add(Stats.Compaction.SUBMITTED_TASKS, numSubmittedCompactionTasks);
- updateCompactionSnapshotStats(currentRunAutoCompactionSnapshotBuilders, iterator, stats);
+ updateCompactionSnapshotStats(compactionSnapshotBuilder, iterator, compactionConfigs);
}
private void resetCompactionSnapshot()
@@ -271,333 +213,168 @@ private boolean isCompactionSupervisorEnabled()
}
}
- /**
- * Queries the Overlord for the status of all tasks that were submitted
- * recently but are not active anymore. The statuses are then updated in the
- * {@link #statusTracker}.
- */
- private void trackStatusOfCompletedTasks(Set activeTaskIds)
- {
- final Set finishedTaskIds = new HashSet<>(statusTracker.getSubmittedTaskIds());
- finishedTaskIds.removeAll(activeTaskIds);
-
- if (finishedTaskIds.isEmpty()) {
- return;
- }
-
- final Map taskStatusMap
- = FutureUtils.getUnchecked(overlordClient.taskStatuses(finishedTaskIds), true);
- for (String taskId : finishedTaskIds) {
- // Assume unknown task to have finished successfully
- final TaskStatus taskStatus = taskStatusMap.getOrDefault(taskId, TaskStatus.success(taskId));
- if (taskStatus.isComplete()) {
- statusTracker.onTaskFinished(taskId, taskStatus);
- }
- }
- }
-
- /**
- * Cancels a currently running compaction task if the segment granularity
- * for this datasource has changed in the compaction config.
- *
- * @return true if the task was canceled, false otherwise.
- */
- private boolean cancelTaskIfGranularityChanged(
- ClientCompactionTaskQuery compactionTaskQuery,
- DataSourceCompactionConfig dataSourceCompactionConfig
- )
- {
- if (dataSourceCompactionConfig == null
- || dataSourceCompactionConfig.getGranularitySpec() == null
- || compactionTaskQuery.getGranularitySpec() == null) {
- return false;
- }
-
- Granularity configuredSegmentGranularity = dataSourceCompactionConfig.getGranularitySpec()
- .getSegmentGranularity();
- Granularity taskSegmentGranularity = compactionTaskQuery.getGranularitySpec().getSegmentGranularity();
- if (configuredSegmentGranularity == null || configuredSegmentGranularity.equals(taskSegmentGranularity)) {
- return false;
- }
-
- LOG.info(
- "Cancelling task[%s] as task segmentGranularity[%s] differs from compaction config segmentGranularity[%s].",
- compactionTaskQuery.getId(), taskSegmentGranularity, configuredSegmentGranularity
- );
- overlordClient.cancelTask(compactionTaskQuery.getId());
- return true;
- }
-
- /**
- * Gets a List of Intervals locked by higher priority tasks for each datasource.
- * However, when using a REPLACE lock for compaction, intervals locked with any APPEND lock will not be returned
- * Since compaction tasks submitted for these Intervals would have to wait anyway,
- * we skip these Intervals until the next compaction run.
- *
- * For now, Segment Locks are being treated the same as Time Chunk Locks even
- * though they lock only a Segment and not the entire Interval. Thus,
- * a compaction task will not be submitted for an Interval if
- *
- * - either the whole Interval is locked by a higher priority Task with an incompatible lock type
- * - or there is atleast one Segment in the Interval that is locked by a
- * higher priority Task
- *
- */
- private Map> getLockedIntervals(
- List compactionConfigs
- )
- {
- final List lockFilterPolicies = compactionConfigs
- .stream()
- .map(config ->
- new LockFilterPolicy(config.getDataSource(), config.getTaskPriority(), null, config.getTaskContext()))
- .collect(Collectors.toList());
- final Map> datasourceToLockedIntervals =
- new HashMap<>(FutureUtils.getUnchecked(overlordClient.findLockedIntervals(lockFilterPolicies), true));
- LOG.debug(
- "Skipping the following intervals for Compaction as they are currently locked: %s",
- datasourceToLockedIntervals
- );
-
- return datasourceToLockedIntervals;
- }
-
- /**
- * Returns the maximum number of task slots used by one native compaction task at any time when the task is
- * issued with the given tuningConfig.
- */
- public static int findMaxNumTaskSlotsUsedByOneNativeCompactionTask(
- @Nullable ClientCompactionTaskQueryTuningConfig tuningConfig
- )
- {
- if (isParallelMode(tuningConfig)) {
- @Nullable
- Integer maxNumConcurrentSubTasks = tuningConfig.getMaxNumConcurrentSubTasks();
- // Max number of task slots used in parallel mode = maxNumConcurrentSubTasks + 1 (supervisor task)
- return (maxNumConcurrentSubTasks == null ? 1 : maxNumConcurrentSubTasks) + 1;
- } else {
- return 1;
- }
- }
-
- /**
- * Returns the maximum number of task slots used by one MSQ compaction task at any time when the task is
- * issued with the given context.
- */
- static int findMaxNumTaskSlotsUsedByOneMsqCompactionTask(@Nullable Map context)
- {
- return context == null
- ? ClientMSQContext.DEFAULT_MAX_NUM_TASKS
- : (int) context.getOrDefault(ClientMSQContext.CTX_MAX_NUM_TASKS, ClientMSQContext.DEFAULT_MAX_NUM_TASKS);
- }
-
-
- /**
- * Returns true if the compaction task can run in the parallel mode with the given tuningConfig.
- * This method should be synchronized with ParallelIndexSupervisorTask.isParallelMode(InputSource, ParallelIndexTuningConfig).
- */
- @VisibleForTesting
- static boolean isParallelMode(@Nullable ClientCompactionTaskQueryTuningConfig tuningConfig)
- {
- if (null == tuningConfig) {
- return false;
- }
- boolean useRangePartitions = useRangePartitions(tuningConfig);
- int minRequiredNumConcurrentSubTasks = useRangePartitions ? 1 : 2;
- return tuningConfig.getMaxNumConcurrentSubTasks() != null
- && tuningConfig.getMaxNumConcurrentSubTasks() >= minRequiredNumConcurrentSubTasks;
- }
-
- private static boolean useRangePartitions(ClientCompactionTaskQueryTuningConfig tuningConfig)
- {
- // dynamic partitionsSpec will be used if getPartitionsSpec() returns null
- return tuningConfig.getPartitionsSpec() instanceof DimensionRangePartitionsSpec;
- }
-
- private int getCompactionTaskCapacity(DruidCompactionConfig dynamicConfig)
- {
- int totalWorkerCapacity = CoordinatorDutyUtils.getTotalWorkerCapacity(overlordClient);
-
- return Math.min(
- (int) (totalWorkerCapacity * dynamicConfig.getCompactionTaskSlotRatio()),
- dynamicConfig.getMaxCompactionTaskSlots()
- );
- }
-
- private int getAvailableCompactionTaskSlots(int compactionTaskCapacity, int busyCompactionTaskSlots)
- {
- final int availableCompactionTaskSlots;
- if (busyCompactionTaskSlots > 0) {
- availableCompactionTaskSlots = Math.max(0, compactionTaskCapacity - busyCompactionTaskSlots);
- } else {
- // compactionTaskCapacity might be 0 if totalWorkerCapacity is low.
- // This guarantees that at least one slot is available if
- // compaction is enabled and estimatedIncompleteCompactionTasks is 0.
- availableCompactionTaskSlots = Math.max(1, compactionTaskCapacity);
- }
- LOG.debug(
- "Found [%d] available task slots for compaction out of max compaction task capacity [%d]",
- availableCompactionTaskSlots, compactionTaskCapacity
- );
-
- return availableCompactionTaskSlots;
- }
-
/**
* Submits compaction tasks to the Overlord. Returns total number of tasks submitted.
*/
private int submitCompactionTasks(
Map compactionConfigs,
- Map currentRunAutoCompactionSnapshotBuilders,
- int numAvailableCompactionTaskSlots,
+ CompactionSnapshotBuilder snapshotBuilder,
+ CompactionSlotManager slotManager,
CompactionSegmentIterator iterator,
+ CompactionCandidateSearchPolicy policy,
CompactionEngine defaultEngine
)
{
- if (numAvailableCompactionTaskSlots <= 0) {
+ if (slotManager.getNumAvailableTaskSlots() <= 0) {
return 0;
}
int numSubmittedTasks = 0;
int totalTaskSlotsAssigned = 0;
- while (iterator.hasNext() && totalTaskSlotsAssigned < numAvailableCompactionTaskSlots) {
+ while (iterator.hasNext() && totalTaskSlotsAssigned < slotManager.getNumAvailableTaskSlots()) {
final CompactionCandidate entry = iterator.next();
final String dataSourceName = entry.getDataSource();
+ final DataSourceCompactionConfig config = compactionConfigs.get(dataSourceName);
- // As these segments will be compacted, we will aggregate the statistic to the Compacted statistics
- currentRunAutoCompactionSnapshotBuilders
- .computeIfAbsent(dataSourceName, AutoCompactionSnapshot::builder)
- .incrementCompactedStats(entry.getStats());
+ final CompactionStatus compactionStatus =
+ statusTracker.computeCompactionStatus(entry, policy);
+ final CompactionCandidate candidatesWithStatus = entry.withCurrentStatus(compactionStatus);
+ statusTracker.onCompactionStatusComputed(candidatesWithStatus, config);
- final DataSourceCompactionConfig config = compactionConfigs.get(dataSourceName);
- final List segmentsToCompact = entry.getSegments();
-
- // Create granularitySpec to send to compaction task
- Granularity segmentGranularityToUse = null;
- if (config.getGranularitySpec() == null || config.getGranularitySpec().getSegmentGranularity() == null) {
- // Determines segmentGranularity from the segmentsToCompact
- // Each batch of segmentToCompact from CompactionSegmentIterator will contain the same interval as
- // segmentGranularity is not set in the compaction config
- Interval interval = segmentsToCompact.get(0).getInterval();
- if (segmentsToCompact.stream().allMatch(segment -> interval.overlaps(segment.getInterval()))) {
- try {
- segmentGranularityToUse = GranularityType.fromPeriod(interval.toPeriod()).getDefaultGranularity();
- }
- catch (IllegalArgumentException iae) {
- // This case can happen if the existing segment interval result in complicated periods.
- // Fall back to setting segmentGranularity as null
- LOG.warn("Cannot determine segmentGranularity from interval[%s].", interval);
- }
- } else {
- LOG.warn(
- "Not setting 'segmentGranularity' for auto-compaction task as"
- + " the segments to compact do not have the same interval."
- );
- }
+ if (compactionStatus.isComplete()) {
+ snapshotBuilder.addToComplete(candidatesWithStatus);
+ } else if (compactionStatus.isSkipped()) {
+ snapshotBuilder.addToSkipped(candidatesWithStatus);
} else {
- segmentGranularityToUse = config.getGranularitySpec().getSegmentGranularity();
+ // As these segments will be compacted, we will aggregate the statistic to the Compacted statistics
+ snapshotBuilder.addToComplete(entry);
}
- final ClientCompactionTaskGranularitySpec granularitySpec = new ClientCompactionTaskGranularitySpec(
- segmentGranularityToUse,
- config.getGranularitySpec() != null ? config.getGranularitySpec().getQueryGranularity() : null,
- config.getGranularitySpec() != null ? config.getGranularitySpec().isRollup() : null
- );
- // Create dimensionsSpec to send to compaction task
- final ClientCompactionTaskDimensionsSpec dimensionsSpec;
- if (config.getDimensionsSpec() != null) {
- dimensionsSpec = new ClientCompactionTaskDimensionsSpec(
- config.getDimensionsSpec().getDimensions()
- );
- } else {
- dimensionsSpec = null;
- }
+ final ClientCompactionTaskQuery taskPayload = createCompactionTask(entry, config, defaultEngine);
- Boolean dropExisting = null;
- if (config.getIoConfig() != null) {
- dropExisting = config.getIoConfig().isDropExisting();
- }
+ final String taskId = taskPayload.getId();
+ FutureUtils.getUnchecked(overlordClient.runTask(taskId, taskPayload), true);
+ statusTracker.onTaskSubmitted(taskId, entry);
- // If all the segments found to be compacted are tombstones then dropExisting
- // needs to be forced to true. This forcing needs to happen in the case that
- // the flag is null, or it is false. It is needed when it is null to avoid the
- // possibility of the code deciding to default it to false later.
- // Forcing the flag to true will enable the task ingestion code to generate new, compacted, tombstones to
- // cover the tombstones found to be compacted as well as to mark them
- // as compacted (update their lastCompactionState). If we don't force the
- // flag then every time this compact duty runs it will find the same tombstones
- // in the interval since their lastCompactionState
- // was not set repeating this over and over and the duty will not make progress; it
- // will become stuck on this set of tombstones.
- // This forcing code should be revised
- // when/if the autocompaction code policy to decide which segments to compact changes
- if (dropExisting == null || !dropExisting) {
- if (segmentsToCompact.stream().allMatch(DataSegment::isTombstone)) {
- dropExisting = true;
- LOG.info("Forcing dropExisting to true since all segments to compact are tombstones.");
- }
- }
+ LOG.debug(
+ "Submitted a compaction task[%s] for [%d] segments in datasource[%s], umbrella interval[%s].",
+ taskId, entry.numSegments(), dataSourceName, entry.getUmbrellaInterval()
+ );
+ LOG.debugSegments(entry.getSegments(), "Compacting segments");
+ numSubmittedTasks++;
+ totalTaskSlotsAssigned += slotManager.computeSlotsRequiredForTask(taskPayload, config);
+ }
- final CompactionEngine compactionEngine = config.getEngine() == null ? defaultEngine : config.getEngine();
- final Map autoCompactionContext = newAutoCompactionContext(config.getTaskContext());
- int slotsRequiredForCurrentTask;
-
- if (compactionEngine == CompactionEngine.MSQ) {
- if (autoCompactionContext.containsKey(ClientMSQContext.CTX_MAX_NUM_TASKS)) {
- slotsRequiredForCurrentTask = (int) autoCompactionContext.get(ClientMSQContext.CTX_MAX_NUM_TASKS);
- } else {
- // Since MSQ needs all task slots for the calculated #tasks to be available upfront, allot all available
- // compaction slots (upto a max of MAX_TASK_SLOTS_FOR_MSQ_COMPACTION) to current compaction task to avoid
- // stalling. Setting "taskAssignment" to "auto" has the problem of not being able to determine the actual
- // count, which is required for subsequent tasks.
- slotsRequiredForCurrentTask = Math.min(
- // Update the slots to 2 (min required for MSQ) if only 1 slot is available.
- numAvailableCompactionTaskSlots == 1 ? 2 : numAvailableCompactionTaskSlots,
- ClientMSQContext.MAX_TASK_SLOTS_FOR_MSQ_COMPACTION_TASK
- );
- autoCompactionContext.put(ClientMSQContext.CTX_MAX_NUM_TASKS, slotsRequiredForCurrentTask);
+ LOG.info("Submitted a total of [%d] compaction tasks.", numSubmittedTasks);
+ return numSubmittedTasks;
+ }
+
+ /**
+ * Creates a {@link ClientCompactionTaskQuery} which can be submitted to an
+ * {@link OverlordClient} to start a compaction task.
+ */
+ public static ClientCompactionTaskQuery createCompactionTask(
+ CompactionCandidate candidate,
+ DataSourceCompactionConfig config,
+ CompactionEngine defaultEngine
+ )
+ {
+ final List segmentsToCompact = candidate.getSegments();
+
+ // Create granularitySpec to send to compaction task
+ Granularity segmentGranularityToUse = null;
+ if (config.getGranularitySpec() == null || config.getGranularitySpec().getSegmentGranularity() == null) {
+ // Determines segmentGranularity from the segmentsToCompact
+ // Each batch of segmentToCompact from CompactionSegmentIterator will contain the same interval as
+ // segmentGranularity is not set in the compaction config
+ Interval interval = segmentsToCompact.get(0).getInterval();
+ if (segmentsToCompact.stream().allMatch(segment -> interval.overlaps(segment.getInterval()))) {
+ try {
+ segmentGranularityToUse = GranularityType.fromPeriod(interval.toPeriod()).getDefaultGranularity();
+ }
+ catch (IllegalArgumentException iae) {
+ // This case can happen if the existing segment interval result in complicated periods.
+ // Fall back to setting segmentGranularity as null
+ LOG.warn("Cannot determine segmentGranularity from interval[%s].", interval);
}
} else {
- slotsRequiredForCurrentTask = findMaxNumTaskSlotsUsedByOneNativeCompactionTask(config.getTuningConfig());
+ LOG.warn(
+ "Not setting 'segmentGranularity' for auto-compaction task as"
+ + " the segments to compact do not have the same interval."
+ );
}
+ } else {
+ segmentGranularityToUse = config.getGranularitySpec().getSegmentGranularity();
+ }
+ final ClientCompactionTaskGranularitySpec granularitySpec = new ClientCompactionTaskGranularitySpec(
+ segmentGranularityToUse,
+ config.getGranularitySpec() != null ? config.getGranularitySpec().getQueryGranularity() : null,
+ config.getGranularitySpec() != null ? config.getGranularitySpec().isRollup() : null
+ );
+
+ // Create dimensionsSpec to send to compaction task
+ final ClientCompactionTaskDimensionsSpec dimensionsSpec;
+ if (config.getDimensionsSpec() != null) {
+ dimensionsSpec = new ClientCompactionTaskDimensionsSpec(
+ config.getDimensionsSpec().getDimensions()
+ );
+ } else {
+ dimensionsSpec = null;
+ }
- if (entry.getCurrentStatus() != null) {
- autoCompactionContext.put(COMPACTION_REASON_KEY, entry.getCurrentStatus().getReason());
+ Boolean dropExisting = null;
+ if (config.getIoConfig() != null) {
+ dropExisting = config.getIoConfig().isDropExisting();
+ }
+
+ // If all the segments found to be compacted are tombstones then dropExisting
+ // needs to be forced to true. This forcing needs to happen in the case that
+ // the flag is null, or it is false. It is needed when it is null to avoid the
+ // possibility of the code deciding to default it to false later.
+ // Forcing the flag to true will enable the task ingestion code to generate new, compacted, tombstones to
+ // cover the tombstones found to be compacted as well as to mark them
+ // as compacted (update their lastCompactionState). If we don't force the
+ // flag then every time this compact duty runs it will find the same tombstones
+ // in the interval since their lastCompactionState
+ // was not set repeating this over and over and the duty will not make progress; it
+ // will become stuck on this set of tombstones.
+ // This forcing code should be revised
+ // when/if the autocompaction code policy to decide which segments to compact changes
+ if (dropExisting == null || !dropExisting) {
+ if (segmentsToCompact.stream().allMatch(DataSegment::isTombstone)) {
+ dropExisting = true;
+ LOG.info("Forcing dropExisting to true since all segments to compact are tombstones.");
}
+ }
- final String taskId = compactSegments(
- entry,
- config.getTaskPriority(),
- ClientCompactionTaskQueryTuningConfig.from(
- config.getTuningConfig(),
- config.getMaxRowsPerSegment(),
- config.getMetricsSpec() != null
- ),
- granularitySpec,
- dimensionsSpec,
- config.getMetricsSpec(),
- config.getTransformSpec(),
- config.getProjections(),
- dropExisting,
- autoCompactionContext,
- new ClientCompactionRunnerInfo(compactionEngine)
- );
+ final CompactionEngine compactionEngine = config.getEngine() == null ? defaultEngine : config.getEngine();
+ final Map autoCompactionContext = newAutoCompactionContext(config.getTaskContext());
- LOG.debug(
- "Submitted a compaction task[%s] for [%d] segments in datasource[%s], umbrella interval[%s].",
- taskId, segmentsToCompact.size(), dataSourceName, entry.getUmbrellaInterval()
- );
- LOG.debugSegments(segmentsToCompact, "Compacting segments");
- numSubmittedTasks++;
- totalTaskSlotsAssigned += slotsRequiredForCurrentTask;
+ if (candidate.getCurrentStatus() != null) {
+ autoCompactionContext.put(COMPACTION_REASON_KEY, candidate.getCurrentStatus().getReason());
}
- LOG.info("Submitted a total of [%d] compaction tasks.", numSubmittedTasks);
- return numSubmittedTasks;
+ return compactSegments(
+ candidate,
+ config.getTaskPriority(),
+ ClientCompactionTaskQueryTuningConfig.from(
+ config.getTuningConfig(),
+ config.getMaxRowsPerSegment(),
+ config.getMetricsSpec() != null
+ ),
+ granularitySpec,
+ dimensionsSpec,
+ config.getMetricsSpec(),
+ config.getTransformSpec(),
+ config.getProjections(),
+ dropExisting,
+ autoCompactionContext,
+ new ClientCompactionRunnerInfo(compactionEngine)
+ );
}
- private Map newAutoCompactionContext(@Nullable Map configuredContext)
+ private static Map newAutoCompactionContext(@Nullable Map configuredContext)
{
final Map newContext = configuredContext == null
? new HashMap<>()
@@ -607,60 +384,23 @@ private Map newAutoCompactionContext(@Nullable Map currentRunAutoCompactionSnapshotBuilders,
+ CompactionSnapshotBuilder snapshotBuilder,
CompactionSegmentIterator iterator,
- CoordinatorRunStats stats
+ Map datasourceToConfig
)
{
// Mark all the segments remaining in the iterator as "awaiting compaction"
while (iterator.hasNext()) {
- final CompactionCandidate entry = iterator.next();
- currentRunAutoCompactionSnapshotBuilders
- .computeIfAbsent(entry.getDataSource(), AutoCompactionSnapshot::builder)
- .incrementWaitingStats(entry.getStats());
+ snapshotBuilder.addToPending(iterator.next());
}
-
- // Statistics of all segments considered compacted after this run
- iterator.getCompactedSegments().forEach(
- candidateSegments -> currentRunAutoCompactionSnapshotBuilders
- .computeIfAbsent(candidateSegments.getDataSource(), AutoCompactionSnapshot::builder)
- .incrementCompactedStats(candidateSegments.getStats())
- );
-
- // Statistics of all segments considered skipped after this run
- iterator.getSkippedSegments().forEach(
- candidateSegments -> currentRunAutoCompactionSnapshotBuilders
- .computeIfAbsent(candidateSegments.getDataSource(), AutoCompactionSnapshot::builder)
- .incrementSkippedStats(candidateSegments.getStats())
- );
-
- final Map currentAutoCompactionSnapshotPerDataSource = new HashMap<>();
- currentRunAutoCompactionSnapshotBuilders.forEach((dataSource, builder) -> {
- final AutoCompactionSnapshot autoCompactionSnapshot = builder.build();
- currentAutoCompactionSnapshotPerDataSource.put(dataSource, autoCompactionSnapshot);
- collectSnapshotStats(autoCompactionSnapshot, stats);
+ iterator.getCompactedSegments().forEach(snapshotBuilder::addToComplete);
+ iterator.getSkippedSegments().forEach(entry -> {
+ statusTracker.onCompactionStatusComputed(entry, datasourceToConfig.get(entry.getDataSource()));
+ snapshotBuilder.addToSkipped(entry);
});
// Atomic update of autoCompactionSnapshotPerDataSource with the latest from this coordinator run
- autoCompactionSnapshotPerDataSource.set(currentAutoCompactionSnapshotPerDataSource);
- }
-
- private void collectSnapshotStats(
- AutoCompactionSnapshot autoCompactionSnapshot,
- CoordinatorRunStats stats
- )
- {
- final RowKey rowKey = RowKey.of(Dimension.DATASOURCE, autoCompactionSnapshot.getDataSource());
-
- stats.add(Stats.Compaction.PENDING_BYTES, rowKey, autoCompactionSnapshot.getBytesAwaitingCompaction());
- stats.add(Stats.Compaction.PENDING_SEGMENTS, rowKey, autoCompactionSnapshot.getSegmentCountAwaitingCompaction());
- stats.add(Stats.Compaction.PENDING_INTERVALS, rowKey, autoCompactionSnapshot.getIntervalCountAwaitingCompaction());
- stats.add(Stats.Compaction.COMPACTED_BYTES, rowKey, autoCompactionSnapshot.getBytesCompacted());
- stats.add(Stats.Compaction.COMPACTED_SEGMENTS, rowKey, autoCompactionSnapshot.getSegmentCountCompacted());
- stats.add(Stats.Compaction.COMPACTED_INTERVALS, rowKey, autoCompactionSnapshot.getIntervalCountCompacted());
- stats.add(Stats.Compaction.SKIPPED_BYTES, rowKey, autoCompactionSnapshot.getBytesSkipped());
- stats.add(Stats.Compaction.SKIPPED_SEGMENTS, rowKey, autoCompactionSnapshot.getSegmentCountSkipped());
- stats.add(Stats.Compaction.SKIPPED_INTERVALS, rowKey, autoCompactionSnapshot.getIntervalCountSkipped());
+ autoCompactionSnapshotPerDataSource.set(snapshotBuilder.build());
}
@Nullable
@@ -674,7 +414,7 @@ public Map getAutoCompactionSnapshot()
return autoCompactionSnapshotPerDataSource.get();
}
- private String compactSegments(
+ private static ClientCompactionTaskQuery compactSegments(
CompactionCandidate entry,
int compactionTaskPriority,
ClientCompactionTaskQueryTuningConfig tuningConfig,
@@ -703,7 +443,7 @@ private String compactSegments(
final String taskId = IdUtils.newTaskId(TASK_ID_PREFIX, ClientCompactionTaskQuery.TYPE, dataSource, null);
final Granularity segmentGranularity = granularitySpec == null ? null : granularitySpec.getSegmentGranularity();
- final ClientCompactionTaskQuery taskPayload = new ClientCompactionTaskQuery(
+ return new ClientCompactionTaskQuery(
taskId,
dataSource,
new ClientCompactionIOConfig(
@@ -719,9 +459,5 @@ private String compactSegments(
context,
compactionRunner
);
- FutureUtils.getUnchecked(overlordClient.runTask(taskId, taskPayload), true);
- statusTracker.onTaskSubmitted(taskPayload, entry);
-
- return taskId;
}
}
diff --git a/server/src/main/java/org/apache/druid/server/coordinator/stats/CoordinatorRunStats.java b/server/src/main/java/org/apache/druid/server/coordinator/stats/CoordinatorRunStats.java
index 76d788cb4306..a9473b298cc9 100644
--- a/server/src/main/java/org/apache/druid/server/coordinator/stats/CoordinatorRunStats.java
+++ b/server/src/main/java/org/apache/druid/server/coordinator/stats/CoordinatorRunStats.java
@@ -95,6 +95,20 @@ public long get(CoordinatorStat stat, RowKey rowKey)
return statValues == null ? 0 : statValues.getLong(stat);
}
+ /**
+ * Invokes the {@link StatHandler} for each value of the specified stat.
+ */
+ public void forEachEntry(CoordinatorStat stat, StatHandler handler)
+ {
+ allStats.forEach(
+ (rowKey, stats) -> stats.object2LongEntrySet().fastForEach(entry -> {
+ if (entry.getKey().equals(stat)) {
+ handler.handle(stat, rowKey, entry.getLongValue());
+ }
+ })
+ );
+ }
+
public void forEachStat(StatHandler handler)
{
allStats.forEach(
diff --git a/server/src/main/java/org/apache/druid/server/coordinator/stats/Stats.java b/server/src/main/java/org/apache/druid/server/coordinator/stats/Stats.java
index f4ae1b5f8a9e..32d4af8806ab 100644
--- a/server/src/main/java/org/apache/druid/server/coordinator/stats/Stats.java
+++ b/server/src/main/java/org/apache/druid/server/coordinator/stats/Stats.java
@@ -112,6 +112,8 @@ public static class Compaction
{
public static final CoordinatorStat SUBMITTED_TASKS
= CoordinatorStat.toDebugAndEmit("compactTasks", "compact/task/count");
+ public static final CoordinatorStat CANCELLED_TASKS
+ = CoordinatorStat.toDebugAndEmit("compactCancelled", "compactTask/cancelled/count");
public static final CoordinatorStat MAX_SLOTS
= CoordinatorStat.toDebugAndEmit("compactMaxSlots", "compactTask/maxSlot/count");
public static final CoordinatorStat AVAILABLE_SLOTS
diff --git a/server/src/test/java/org/apache/druid/server/compaction/CompactionRunSimulatorTest.java b/server/src/test/java/org/apache/druid/server/compaction/CompactionRunSimulatorTest.java
index fd491ef930f8..bab2e351df11 100644
--- a/server/src/test/java/org/apache/druid/server/compaction/CompactionRunSimulatorTest.java
+++ b/server/src/test/java/org/apache/druid/server/compaction/CompactionRunSimulatorTest.java
@@ -19,14 +19,12 @@
package org.apache.druid.server.compaction;
-import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import org.apache.druid.client.indexing.TaskPayloadResponse;
import org.apache.druid.indexer.CompactionEngine;
import org.apache.druid.indexer.TaskStatus;
import org.apache.druid.indexer.TaskStatusPlus;
-import org.apache.druid.jackson.DefaultObjectMapper;
import org.apache.druid.java.util.common.CloseableIterators;
import org.apache.druid.java.util.common.Intervals;
import org.apache.druid.java.util.common.granularity.Granularities;
@@ -51,10 +49,8 @@
public class CompactionRunSimulatorTest
{
- private static final ObjectMapper OBJECT_MAPPER = new DefaultObjectMapper();
-
private final CompactionRunSimulator simulator = new CompactionRunSimulator(
- new CompactionStatusTracker(OBJECT_MAPPER),
+ new CompactionStatusTracker(),
new TestOverlordClient()
);
@@ -115,7 +111,7 @@ public void testSimulateClusterCompactionConfigUpdate()
);
Assert.assertEquals(
Collections.singletonList(
- Arrays.asList("wiki", Intervals.of("2013-01-10/P1D"), 10, 1_000_000_000L, "skip offset from latest[P1D]")
+ Arrays.asList("wiki", Intervals.of("2013-01-10/P1D"), 10, 1_000_000_000L, 1, "skip offset from latest[P1D]")
),
skippedTable.getRows()
);
diff --git a/server/src/test/java/org/apache/druid/server/compaction/CompactionStatusTest.java b/server/src/test/java/org/apache/druid/server/compaction/CompactionStatusTest.java
index a8e6977dc42c..9af152a5cc46 100644
--- a/server/src/test/java/org/apache/druid/server/compaction/CompactionStatusTest.java
+++ b/server/src/test/java/org/apache/druid/server/compaction/CompactionStatusTest.java
@@ -320,8 +320,7 @@ public void testStatusWhenLastCompactionStateSameAsRequired()
final DataSegment segment = DataSegment.builder(WIKI_SEGMENT).lastCompactionState(lastCompactionState).build();
final CompactionStatus status = CompactionStatus.compute(
CompactionCandidate.from(Collections.singletonList(segment)),
- compactionConfig,
- OBJECT_MAPPER
+ compactionConfig
);
Assert.assertTrue(status.isComplete());
}
@@ -370,8 +369,7 @@ public void testStatusWhenProjectionsMatch()
final DataSegment segment = DataSegment.builder(WIKI_SEGMENT).lastCompactionState(lastCompactionState).build();
final CompactionStatus status = CompactionStatus.compute(
CompactionCandidate.from(Collections.singletonList(segment)),
- compactionConfig,
- OBJECT_MAPPER
+ compactionConfig
);
Assert.assertTrue(status.isComplete());
}
@@ -425,8 +423,7 @@ public void testStatusWhenProjectionsMismatch()
final DataSegment segment = DataSegment.builder(WIKI_SEGMENT).lastCompactionState(lastCompactionState).build();
final CompactionStatus status = CompactionStatus.compute(
CompactionCandidate.from(Collections.singletonList(segment)),
- compactionConfig,
- OBJECT_MAPPER
+ compactionConfig
);
Assert.assertFalse(status.isComplete());
}
@@ -443,8 +440,7 @@ private void verifyCompactionStatusIsPendingBecause(
.build();
final CompactionStatus status = CompactionStatus.compute(
CompactionCandidate.from(Collections.singletonList(segment)),
- compactionConfig,
- OBJECT_MAPPER
+ compactionConfig
);
Assert.assertFalse(status.isComplete());
diff --git a/server/src/test/java/org/apache/druid/server/compaction/CompactionStatusTrackerTest.java b/server/src/test/java/org/apache/druid/server/compaction/CompactionStatusTrackerTest.java
index 1878aab8c0c3..40a2c501ca02 100644
--- a/server/src/test/java/org/apache/druid/server/compaction/CompactionStatusTrackerTest.java
+++ b/server/src/test/java/org/apache/druid/server/compaction/CompactionStatusTrackerTest.java
@@ -19,16 +19,11 @@
package org.apache.druid.server.compaction;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import org.apache.druid.client.indexing.ClientCompactionTaskQuery;
import org.apache.druid.indexer.TaskState;
import org.apache.druid.indexer.TaskStatus;
-import org.apache.druid.jackson.DefaultObjectMapper;
import org.apache.druid.java.util.common.DateTimes;
import org.apache.druid.segment.TestDataSource;
import org.apache.druid.server.coordinator.CreateDataSegments;
-import org.apache.druid.server.coordinator.DataSourceCompactionConfig;
-import org.apache.druid.server.coordinator.InlineSchemaDataSourceCompactionConfig;
import org.apache.druid.timeline.DataSegment;
import org.junit.Assert;
import org.junit.Before;
@@ -38,7 +33,6 @@
public class CompactionStatusTrackerTest
{
- private static final ObjectMapper MAPPER = new DefaultObjectMapper();
private static final DataSegment WIKI_SEGMENT
= CreateDataSegments.ofDatasource(TestDataSource.WIKI).eachOfSizeInMb(100).get(0);
@@ -47,7 +41,7 @@ public class CompactionStatusTrackerTest
@Before
public void setup()
{
- statusTracker = new CompactionStatusTracker(MAPPER);
+ statusTracker = new CompactionStatusTracker();
}
@Test
@@ -55,7 +49,7 @@ public void testGetLatestTaskStatusForSubmittedTask()
{
final CompactionCandidate candidateSegments
= CompactionCandidate.from(Collections.singletonList(WIKI_SEGMENT));
- statusTracker.onTaskSubmitted(createCompactionTask("task1"), candidateSegments);
+ statusTracker.onTaskSubmitted("task1", candidateSegments);
CompactionTaskStatus status = statusTracker.getLatestTaskStatus(candidateSegments);
Assert.assertEquals(TaskState.RUNNING, status.getState());
@@ -66,7 +60,7 @@ public void testGetLatestTaskStatusForSuccessfulTask()
{
final CompactionCandidate candidateSegments
= CompactionCandidate.from(Collections.singletonList(WIKI_SEGMENT));
- statusTracker.onTaskSubmitted(createCompactionTask("task1"), candidateSegments);
+ statusTracker.onTaskSubmitted("task1", candidateSegments);
statusTracker.onTaskFinished("task1", TaskStatus.success("task1"));
CompactionTaskStatus status = statusTracker.getLatestTaskStatus(candidateSegments);
@@ -78,7 +72,7 @@ public void testGetLatestTaskStatusForFailedTask()
{
final CompactionCandidate candidateSegments
= CompactionCandidate.from(Collections.singletonList(WIKI_SEGMENT));
- statusTracker.onTaskSubmitted(createCompactionTask("task1"), candidateSegments);
+ statusTracker.onTaskSubmitted("task1", candidateSegments);
statusTracker.onTaskFinished("task1", TaskStatus.failure("task1", "some failure"));
CompactionTaskStatus status = statusTracker.getLatestTaskStatus(candidateSegments);
@@ -92,10 +86,10 @@ public void testGetLatestTaskStatusForRepeatedlyFailingTask()
final CompactionCandidate candidateSegments
= CompactionCandidate.from(Collections.singletonList(WIKI_SEGMENT));
- statusTracker.onTaskSubmitted(createCompactionTask("task1"), candidateSegments);
+ statusTracker.onTaskSubmitted("task1", candidateSegments);
statusTracker.onTaskFinished("task1", TaskStatus.failure("task1", "some failure"));
- statusTracker.onTaskSubmitted(createCompactionTask("task2"), candidateSegments);
+ statusTracker.onTaskSubmitted("task2", candidateSegments);
CompactionTaskStatus status = statusTracker.getLatestTaskStatus(candidateSegments);
Assert.assertEquals(TaskState.RUNNING, status.getState());
Assert.assertEquals(1, status.getNumConsecutiveFailures());
@@ -110,24 +104,22 @@ public void testGetLatestTaskStatusForRepeatedlyFailingTask()
@Test
public void testComputeCompactionStatusForSuccessfulTask()
{
- final DataSourceCompactionConfig compactionConfig
- = InlineSchemaDataSourceCompactionConfig.builder().forDataSource(TestDataSource.WIKI).build();
final NewestSegmentFirstPolicy policy = new NewestSegmentFirstPolicy(null);
final CompactionCandidate candidateSegments
= CompactionCandidate.from(Collections.singletonList(WIKI_SEGMENT));
// Verify that interval is originally eligible for compaction
CompactionStatus status
- = statusTracker.computeCompactionStatus(candidateSegments, compactionConfig, policy);
+ = statusTracker.computeCompactionStatus(candidateSegments, policy);
Assert.assertEquals(CompactionStatus.State.PENDING, status.getState());
Assert.assertEquals("not compacted yet", status.getReason());
// Verify that interval is skipped for compaction after task has finished
statusTracker.onSegmentTimelineUpdated(DateTimes.nowUtc().minusMinutes(1));
- statusTracker.onTaskSubmitted(createCompactionTask("task1"), candidateSegments);
+ statusTracker.onTaskSubmitted("task1", candidateSegments);
statusTracker.onTaskFinished("task1", TaskStatus.success("task1"));
- status = statusTracker.computeCompactionStatus(candidateSegments, compactionConfig, policy);
+ status = statusTracker.computeCompactionStatus(candidateSegments, policy);
Assert.assertEquals(CompactionStatus.State.SKIPPED, status.getState());
Assert.assertEquals(
"Segment timeline not updated since last compaction task succeeded",
@@ -136,26 +128,7 @@ public void testComputeCompactionStatusForSuccessfulTask()
// Verify that interval becomes eligible again after timeline has been updated
statusTracker.onSegmentTimelineUpdated(DateTimes.nowUtc());
- status = statusTracker.computeCompactionStatus(candidateSegments, compactionConfig, policy);
+ status = statusTracker.computeCompactionStatus(candidateSegments, policy);
Assert.assertEquals(CompactionStatus.State.PENDING, status.getState());
}
-
- private ClientCompactionTaskQuery createCompactionTask(
- String taskId
- )
- {
- return new ClientCompactionTaskQuery(
- taskId,
- TestDataSource.WIKI,
- null,
- null,
- null,
- null,
- null,
- null,
- null,
- null,
- null
- );
- }
}
diff --git a/server/src/test/java/org/apache/druid/server/compaction/NewestSegmentFirstPolicyTest.java b/server/src/test/java/org/apache/druid/server/compaction/NewestSegmentFirstPolicyTest.java
index 5cba8cc17c22..dd48bd2aba19 100644
--- a/server/src/test/java/org/apache/druid/server/compaction/NewestSegmentFirstPolicyTest.java
+++ b/server/src/test/java/org/apache/druid/server/compaction/NewestSegmentFirstPolicyTest.java
@@ -65,7 +65,6 @@
import org.joda.time.Interval;
import org.joda.time.Period;
import org.junit.Assert;
-import org.junit.Before;
import org.junit.Test;
import java.util.ArrayList;
@@ -82,13 +81,6 @@ public class NewestSegmentFirstPolicyTest
private static final int DEFAULT_NUM_SEGMENTS_PER_SHARD = 4;
private final ObjectMapper mapper = new DefaultObjectMapper();
private final NewestSegmentFirstPolicy policy = new NewestSegmentFirstPolicy(null);
- private CompactionStatusTracker statusTracker;
-
- @Before
- public void setup()
- {
- statusTracker = new CompactionStatusTracker(mapper);
- }
@Test
public void testLargeOffsetAndSmallSegmentInterval()
@@ -284,8 +276,7 @@ public void testSkipDataSourceWithNoSegments()
.withNumPartitions(4)
)
),
- Collections.emptyMap(),
- statusTracker
+ Collections.emptyMap()
);
assertCompactSegmentIntervals(
@@ -517,8 +508,7 @@ public void testWithSkipIntervals()
Intervals.of("2017-11-15T00:00:00/2017-11-15T20:00:00"),
Intervals.of("2017-11-13T00:00:00/2017-11-14T01:00:00")
)
- ),
- statusTracker
+ )
);
assertCompactSegmentIntervals(
@@ -557,8 +547,7 @@ public void testHoleInSearchInterval()
Intervals.of("2017-11-16T04:00:00/2017-11-16T10:00:00"),
Intervals.of("2017-11-16T14:00:00/2017-11-16T20:00:00")
)
- ),
- statusTracker
+ )
);
assertCompactSegmentIntervals(
@@ -2063,8 +2052,7 @@ TestDataSource.KOALA, configBuilder().forDataSource(TestDataSource.KOALA).build(
TestDataSource.WIKI, SegmentTimeline.forSegments(wikiSegments),
TestDataSource.KOALA, SegmentTimeline.forSegments(koalaSegments)
),
- Collections.emptyMap(),
- statusTracker
+ Collections.emptyMap()
);
// Verify that the segments of WIKI are preferred even though they are older
@@ -2085,8 +2073,7 @@ private CompactionSegmentIterator createIterator(DataSourceCompactionConfig conf
policy,
Collections.singletonMap(TestDataSource.WIKI, config),
Collections.singletonMap(TestDataSource.WIKI, timeline),
- Collections.emptyMap(),
- statusTracker
+ Collections.emptyMap()
);
}
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 f77da3140a4b..39213ff8616d 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
@@ -139,8 +139,7 @@ public void setUp() throws Exception
)
).andReturn(new AtomicReference<>(DruidCompactionConfig.empty())).anyTimes();
EasyMock.replay(configManager);
- final ObjectMapper objectMapper = new DefaultObjectMapper();
- statusTracker = new CompactionStatusTracker(objectMapper);
+ statusTracker = new CompactionStatusTracker();
druidCoordinatorConfig = new DruidCoordinatorConfig(
new CoordinatorRunConfig(new Duration(COORDINATOR_START_DELAY), new Duration(COORDINATOR_PERIOD)),
new CoordinatorPeriodConfig(null, null),
@@ -169,7 +168,7 @@ public void setUp() throws Exception
new TestDruidLeaderSelector(),
null,
CentralizedDatasourceSchemaConfig.create(),
- new CompactionStatusTracker(OBJECT_MAPPER),
+ new CompactionStatusTracker(),
EasyMock.niceMock(CoordinatorDynamicConfigSyncer.class),
EasyMock.niceMock(CloneStatusManager.class)
);
@@ -481,7 +480,7 @@ public void testCompactSegmentsDutyWhenCustomDutyGroupEmpty()
new TestDruidLeaderSelector(),
null,
CentralizedDatasourceSchemaConfig.create(),
- new CompactionStatusTracker(OBJECT_MAPPER),
+ new CompactionStatusTracker(),
EasyMock.niceMock(CoordinatorDynamicConfigSyncer.class),
EasyMock.niceMock(CloneStatusManager.class)
);
@@ -533,7 +532,7 @@ public void testInitializeCompactSegmentsDutyWhenCustomDutyGroupDoesNotContainsC
new TestDruidLeaderSelector(),
null,
CentralizedDatasourceSchemaConfig.create(),
- new CompactionStatusTracker(OBJECT_MAPPER),
+ new CompactionStatusTracker(),
EasyMock.niceMock(CoordinatorDynamicConfigSyncer.class),
EasyMock.niceMock(CloneStatusManager.class)
);
@@ -585,7 +584,7 @@ public void testInitializeCompactSegmentsDutyWhenCustomDutyGroupContainsCompactS
new TestDruidLeaderSelector(),
null,
CentralizedDatasourceSchemaConfig.create(),
- new CompactionStatusTracker(OBJECT_MAPPER),
+ new CompactionStatusTracker(),
EasyMock.niceMock(CoordinatorDynamicConfigSyncer.class),
EasyMock.niceMock(CloneStatusManager.class)
);
@@ -695,7 +694,7 @@ public void testCoordinatorCustomDutyGroupsRunAsExpected() throws Exception
new TestDruidLeaderSelector(),
null,
CentralizedDatasourceSchemaConfig.create(),
- new CompactionStatusTracker(OBJECT_MAPPER),
+ new CompactionStatusTracker(),
EasyMock.niceMock(CoordinatorDynamicConfigSyncer.class),
EasyMock.niceMock(CloneStatusManager.class)
);
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 0d0022896066..d7bbb947d9dc 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
@@ -81,6 +81,7 @@
import org.apache.druid.segment.indexing.BatchIOConfig;
import org.apache.druid.segment.transform.CompactionTransformSpec;
import org.apache.druid.server.compaction.CompactionCandidateSearchPolicy;
+import org.apache.druid.server.compaction.CompactionSlotManager;
import org.apache.druid.server.compaction.CompactionStatusTracker;
import org.apache.druid.server.compaction.FixedIntervalOrderPolicy;
import org.apache.druid.server.compaction.NewestSegmentFirstPolicy;
@@ -226,7 +227,7 @@ public void setup()
}
}
dataSources = DataSourcesSnapshot.fromUsedSegments(allSegments);
- statusTracker = new CompactionStatusTracker(JSON_MAPPER);
+ statusTracker = new CompactionStatusTracker();
policy = new NewestSegmentFirstPolicy(null);
}
@@ -2004,7 +2005,7 @@ public static class StaticUtilsTest
@Test
public void testIsParalleModeNullTuningConfigReturnFalse()
{
- Assert.assertFalse(CompactSegments.isParallelMode(null));
+ Assert.assertFalse(CompactionSlotManager.isParallelMode(null));
}
@Test
@@ -2012,7 +2013,7 @@ public void testIsParallelModeNullPartitionsSpecReturnFalse()
{
ClientCompactionTaskQueryTuningConfig tuningConfig = Mockito.mock(ClientCompactionTaskQueryTuningConfig.class);
Mockito.when(tuningConfig.getPartitionsSpec()).thenReturn(null);
- Assert.assertFalse(CompactSegments.isParallelMode(tuningConfig));
+ Assert.assertFalse(CompactionSlotManager.isParallelMode(tuningConfig));
}
@Test
@@ -2022,13 +2023,13 @@ public void testIsParallelModeNonRangePartitionVaryingMaxNumConcurrentSubTasks()
Mockito.when(tuningConfig.getPartitionsSpec()).thenReturn(Mockito.mock(PartitionsSpec.class));
Mockito.when(tuningConfig.getMaxNumConcurrentSubTasks()).thenReturn(null);
- Assert.assertFalse(CompactSegments.isParallelMode(tuningConfig));
+ Assert.assertFalse(CompactionSlotManager.isParallelMode(tuningConfig));
Mockito.when(tuningConfig.getMaxNumConcurrentSubTasks()).thenReturn(1);
- Assert.assertFalse(CompactSegments.isParallelMode(tuningConfig));
+ Assert.assertFalse(CompactionSlotManager.isParallelMode(tuningConfig));
Mockito.when(tuningConfig.getMaxNumConcurrentSubTasks()).thenReturn(2);
- Assert.assertTrue(CompactSegments.isParallelMode(tuningConfig));
+ Assert.assertTrue(CompactionSlotManager.isParallelMode(tuningConfig));
}
@Test
@@ -2038,13 +2039,13 @@ public void testIsParallelModeRangePartitionVaryingMaxNumConcurrentSubTasks()
Mockito.when(tuningConfig.getPartitionsSpec()).thenReturn(Mockito.mock(SingleDimensionPartitionsSpec.class));
Mockito.when(tuningConfig.getMaxNumConcurrentSubTasks()).thenReturn(null);
- Assert.assertFalse(CompactSegments.isParallelMode(tuningConfig));
+ Assert.assertFalse(CompactionSlotManager.isParallelMode(tuningConfig));
Mockito.when(tuningConfig.getMaxNumConcurrentSubTasks()).thenReturn(1);
- Assert.assertTrue(CompactSegments.isParallelMode(tuningConfig));
+ Assert.assertTrue(CompactionSlotManager.isParallelMode(tuningConfig));
Mockito.when(tuningConfig.getMaxNumConcurrentSubTasks()).thenReturn(2);
- Assert.assertTrue(CompactSegments.isParallelMode(tuningConfig));
+ Assert.assertTrue(CompactionSlotManager.isParallelMode(tuningConfig));
}
@Test
@@ -2053,7 +2054,7 @@ public void testFindMaxNumTaskSlotsUsedByOneCompactionTaskWhenIsParallelMode()
ClientCompactionTaskQueryTuningConfig tuningConfig = Mockito.mock(ClientCompactionTaskQueryTuningConfig.class);
Mockito.when(tuningConfig.getPartitionsSpec()).thenReturn(Mockito.mock(PartitionsSpec.class));
Mockito.when(tuningConfig.getMaxNumConcurrentSubTasks()).thenReturn(2);
- Assert.assertEquals(3, CompactSegments.findMaxNumTaskSlotsUsedByOneNativeCompactionTask(tuningConfig));
+ Assert.assertEquals(3, CompactionSlotManager.getMaxTaskSlotsForNativeCompactionTask(tuningConfig));
}
@Test
@@ -2062,7 +2063,7 @@ public void testFindMaxNumTaskSlotsUsedByOneCompactionTaskWhenIsSequentialMode()
ClientCompactionTaskQueryTuningConfig tuningConfig = Mockito.mock(ClientCompactionTaskQueryTuningConfig.class);
Mockito.when(tuningConfig.getPartitionsSpec()).thenReturn(Mockito.mock(PartitionsSpec.class));
Mockito.when(tuningConfig.getMaxNumConcurrentSubTasks()).thenReturn(1);
- Assert.assertEquals(1, CompactSegments.findMaxNumTaskSlotsUsedByOneNativeCompactionTask(tuningConfig));
+ Assert.assertEquals(1, CompactionSlotManager.getMaxTaskSlotsForNativeCompactionTask(tuningConfig));
}
}
diff --git a/server/src/test/java/org/apache/druid/server/coordinator/simulate/CoordinatorSimulationBuilder.java b/server/src/test/java/org/apache/druid/server/coordinator/simulate/CoordinatorSimulationBuilder.java
index 8d6349237531..0838b41402d5 100644
--- a/server/src/test/java/org/apache/druid/server/coordinator/simulate/CoordinatorSimulationBuilder.java
+++ b/server/src/test/java/org/apache/druid/server/coordinator/simulate/CoordinatorSimulationBuilder.java
@@ -223,7 +223,7 @@ public CoordinatorSimulation build()
env.leaderSelector,
null,
CentralizedDatasourceSchemaConfig.create(),
- new CompactionStatusTracker(OBJECT_MAPPER),
+ new CompactionStatusTracker(),
env.configSyncer,
env.cloneStatusManager
);