This is an automated email from the ASF dual-hosted git repository.
kfaraz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new be27640963c perf: Improve efficiency of embedded kill tasks (#19772)
be27640963c is described below
commit be27640963ce7aae29659146d01b2ec987146c68
Author: Kashif Faraz <[email protected]>
AuthorDate: Fri Aug 7 09:35:43 2026 +0530
perf: Improve efficiency of embedded kill tasks (#19772)
Description
-----------
The `UnusedSegmentsKiller` currenly has some limitations:
- Wasted cycles: The kill queue contains jobs for all intervals which have
an unused segment,
regardless of whether the unused segment is eligible for kill or not.
- Slow progress: For intervals with a large number of killable unused
segments, a single
embedded kill task is launched which would kill only 1000 segments in one
go. In such cases,
the `UnusedSegmentsKiller` would take very long to clear a backlog of
killable unused segments.
- Many metadata queries: When the kill queue is rebuilt, heavy metadata
queries are fired for
each datasource in the DB, even if they don't have any unused segments,
killable or otherwise.
Changes
---------
- While rebuilding the kill queue, fire a single metadata query to retrieve
the intervals containing
killable unused segments of all datasources.
- This avoids firing multiple queries, one for each datasource in the DB.
- It also ensures that every embedded kill task actually kills atleast 1
unused segment.
- In the above query, scan upto a maximum of 200k eligible unused segments.
This would also
be the maximum number of segments killed when the queue has been fully
processed.
- This number is large enough to make meaningful progress in each cycle
of the
`UnusedSegmentsKiller` and small enough to keep the query time within ~5s
(tried on MySQL 8
with 17M unused segments).
- Pass in the max segments to kill in each `KillCandidate`.
- Allow each kill task to kill multiple batches of segments, upto a maximum
of 10 (previously 1).
---
docs/operations/metrics.md | 1 +
.../indexing/common/actions/SegmentNukeAction.java | 13 ++-
.../druid/indexing/common/actions/TaskLocks.java | 56 ++++++++++
.../common/task/KillUnusedSegmentsTask.java | 49 ++++++---
.../overlord/duty/UnusedSegmentsKiller.java | 121 +++++++++++++--------
.../overlord/supervisor/SupervisorManager.java | 22 +---
.../common/task/KillUnusedSegmentsTaskTest.java | 2 +-
.../overlord/duty/UnusedSegmentsKillerTest.java | 107 ++++++++++++------
.../http/OverlordDataSourcesResourceTest.java | 3 +-
.../TestIndexerMetadataStorageCoordinator.java | 11 ++
.../apache/druid/timeline/DatasourceInterval.java | 29 +++++
.../IndexerMetadataStorageCoordinator.java | 11 ++
.../IndexerSQLMetadataStorageCoordinator.java | 13 +++
.../metadata/SegmentsMetadataManagerConfig.java | 2 +-
.../druid/metadata/SqlSegmentsMetadataQuery.java | 95 +++++++++++++++-
.../druid/metadata/UnusedSegmentKillerConfig.java | 39 ++++++-
.../IndexerSQLMetadataStorageCoordinatorTest.java | 35 +++++-
17 files changed, 484 insertions(+), 125 deletions(-)
diff --git a/docs/operations/metrics.md b/docs/operations/metrics.md
index 3943b59f076..c05f6cc91f9 100644
--- a/docs/operations/metrics.md
+++ b/docs/operations/metrics.md
@@ -418,6 +418,7 @@ These metrics are emitted only if [auto-kill of unused
segments](../data-managem
|`segment/killed/metadataStore/count`|Number of segments permanently deleted
from the metadata store.|`taskId`, `groupId`, `taskType`(=`kill`), `dataSource`|
|`segment/killed/deepStorage/count`|Number of segments permanently deleted
from the deep storage.|`taskId`, `groupId`, `taskType`(=`kill`), `dataSource`|
|`segment/kill/unusedIntervals/count`|Number of intervals containing unused
segments for a given datasource.|`dataSource`|
+|`segment/kill/eligibleSegments/count`|Number of unused segments in an
interval that are eligible for kill.|`dataSource`, `interval`|
|`segment/kill/skippedIntervals/count`|Number of intervals that were skipped
for kill due to being already locked by another task.|`taskId`, `groupId`,
`taskType`(=`kill`), `dataSource`|
|`segment/kill/queueReset/time`|Time taken in milliseconds to reset the kill
queue.||
|`segment/kill/queueProcess/time`|Time taken in milliseconds to fully process
the kill queue.||
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/common/actions/SegmentNukeAction.java
b/indexing-service/src/main/java/org/apache/druid/indexing/common/actions/SegmentNukeAction.java
index 6c73f664e3f..b9e6f035a80 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/common/actions/SegmentNukeAction.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/common/actions/SegmentNukeAction.java
@@ -39,8 +39,10 @@ import java.util.stream.Collectors;
/**
* Permanently deletes unused segments from the metadata store.
+ * {@link #perform(Task, TaskActionToolbox)} returns the number of segments
+ * deleted from the metadata store, but older versions return a null value.
*/
-public class SegmentNukeAction implements TaskAction<Void>
+public class SegmentNukeAction implements TaskAction<Integer>
{
private static final Logger log = new Logger(SegmentNukeAction.class);
@@ -61,19 +63,20 @@ public class SegmentNukeAction implements TaskAction<Void>
}
@Override
- public TypeReference<Void> getReturnTypeReference()
+ public TypeReference<Integer> getReturnTypeReference()
{
return new TypeReference<>() {};
}
@Override
- public Void perform(Task task, TaskActionToolbox toolbox)
+ public Integer perform(Task task, TaskActionToolbox toolbox)
{
TaskLocks.checkLockCoversSegments(task, toolbox.getTaskLockbox(),
segments);
+ final int numDeletedSegments;
try {
final Set<Interval> intervals =
segments.stream().map(DataSegment::getInterval).collect(Collectors.toSet());
- int numDeletedSegments = toolbox.getTaskLockbox().doInCriticalSection(
+ numDeletedSegments = toolbox.getTaskLockbox().doInCriticalSection(
task,
intervals,
CriticalAction.<Integer>builder().onValidLocks(
@@ -106,7 +109,7 @@ public class SegmentNukeAction implements TaskAction<Void>
toolbox.getEmitter().emit(metricBuilder.setMetric("segment/nuked/bytes",
segment.getSize()));
}
- return null;
+ return numDeletedSegments;
}
@Override
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/common/actions/TaskLocks.java
b/indexing-service/src/main/java/org/apache/druid/indexing/common/actions/TaskLocks.java
index 82c8990c748..3646d4a9048 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/common/actions/TaskLocks.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/common/actions/TaskLocks.java
@@ -33,6 +33,7 @@ import org.apache.druid.indexing.overlord.GlobalTaskLockbox;
import org.apache.druid.java.util.common.DateTimes;
import org.apache.druid.java.util.common.ISE;
import org.apache.druid.metadata.ReplaceTaskLock;
+import org.apache.druid.query.QueryContexts;
import org.apache.druid.timeline.DataSegment;
import org.joda.time.DateTime;
import org.joda.time.Interval;
@@ -170,6 +171,61 @@ public class TaskLocks
}
}
+ /**
+ * Checks if an append task with the given context should use concurrent
locks.
+ */
+ public static boolean shouldUseConcurrentLocksForAppend(Map<String, Object>
context, boolean defaultValue)
+ {
+ return shouldUseConcurrentLocks(context, defaultValue,
TaskLockType.APPEND);
+ }
+
+ /**
+ * Checks if a replace task with the given context should use concurrent
locks.
+ */
+ public static boolean shouldUseConcurrentLocksForReplace(Map<String, Object>
context, boolean defaultValue)
+ {
+ return shouldUseConcurrentLocks(context, defaultValue,
TaskLockType.REPLACE);
+ }
+
+ /**
+ * Checks if an append or replace task with the given context should use
+ * concurrent locks.
+ *
+ * @param context Context used by the task.
+ * @param defaultValue Default behaviour in case the context does
not
+ * explicitly specify any value for {@code
useConcurrentLocks}
+ * or {@code taskLockType}.
+ * @param validConcurrentLockType Expected concurrent lock type (APPEND or
REPLACE)
+ * for the given task.
+ */
+ private static boolean shouldUseConcurrentLocks(
+ Map<String, Object> context,
+ boolean defaultValue,
+ TaskLockType validConcurrentLockType
+ )
+ {
+ if (context == null) {
+ return defaultValue;
+ }
+ Boolean useConcurrentLocks = QueryContexts.getAsBoolean(
+ Tasks.USE_CONCURRENT_LOCKS,
+ context.get(Tasks.USE_CONCURRENT_LOCKS)
+ );
+ if (useConcurrentLocks != null) {
+ return useConcurrentLocks;
+ }
+ TaskLockType taskLockType = QueryContexts.getAsEnum(
+ Tasks.TASK_LOCK_TYPE,
+ context.get(Tasks.TASK_LOCK_TYPE),
+ TaskLockType.class
+ );
+ if (taskLockType != null) {
+ return taskLockType == validConcurrentLockType;
+ }
+
+ return defaultValue;
+ }
+
/**
* Finds locks of type {@link TaskLockType#REPLACE} for each of the given
segments
* that have an interval completely covering the interval of the respective
segments.
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/KillUnusedSegmentsTask.java
b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/KillUnusedSegmentsTask.java
index 21b9e3f6a79..e31ac044d89 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/KillUnusedSegmentsTask.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/KillUnusedSegmentsTask.java
@@ -47,6 +47,7 @@ import org.apache.druid.indexing.overlord.Segments;
import org.apache.druid.java.util.common.ISE;
import org.apache.druid.java.util.common.StringUtils;
import org.apache.druid.java.util.common.logger.Logger;
+import org.apache.druid.query.QueryContexts;
import org.apache.druid.server.coordination.BroadcastDatasourceLoadingSpec;
import org.apache.druid.server.http.DataSegmentPlus;
import org.apache.druid.server.lookup.cache.LookupLoadingSpec;
@@ -87,10 +88,16 @@ import java.util.stream.Collectors;
* <li> Filter the set of unreferenced segments using load specs from the set
of used segments. </li>
* <li> Kill the filtered set of segments from deep storage. </li>
* </ol>
- * Note: When {@link Tasks#USE_CONCURRENT_LOCKS} is true, keep a large buffer
- * period before killing segments after they have been marked as unused.
- * Otherwise, there may be a potential data loss if a concurrent APPEND job
- * upgrades one of the segments that are being killed.
+ * <p>
+ * <b>Use concurrent locks:</b> A kill task using REPLACE locks (enabled by
+ * {@link Tasks#USE_CONCURRENT_LOCKS}) is affected by other tasks in the
following
+ * manner:
+ * <ul>
+ * <li>Not affected by tasks using EXCLUSIVE, SHARED or another REPLACE lock
+ * since they are mutually exclusive with this REPLACE lock.</li>
+ * <li>Not affected by APPEND tasks since they can only upgrade used segments
+ * and do not modify unused segments.</li>
+ * </ul>
*/
public class KillUnusedSegmentsTask extends AbstractFixedIntervalTask
{
@@ -210,6 +217,7 @@ public class KillUnusedSegmentsTask extends
AbstractFixedIntervalTask
{
// Track stats for reporting
int numSegmentsKilled = 0;
+ int totalSegmentsDeletedFromMetadataStore = 0;
int numBatchesProcessed = 0;
// List unused segments
@@ -294,14 +302,23 @@ public class KillUnusedSegmentsTask extends
AbstractFixedIntervalTask
}
// 3. Nuke all eligible unused segments
- taskActionClient.submit(new SegmentNukeAction(unusedSegments));
- emitMetric(toolbox.getEmitter(),
TaskMetrics.SEGMENTS_DELETED_FROM_METADATA_STORE, unusedIdToSegmentPlus.size());
+ final Number nukeResult = taskActionClient.submit(new
SegmentNukeAction(unusedSegments));
+
+ // Older versions of the Overlord return a null value from the
SegmentNukeAction
+ final int numSegmentsDeletedFromMetadataStore =
+ nukeResult == null ? unusedSegments.size() : nukeResult.intValue();
+ emitMetric(
+ toolbox.getEmitter(),
+ TaskMetrics.SEGMENTS_DELETED_FROM_METADATA_STORE,
+ numSegmentsDeletedFromMetadataStore
+ );
// 4. Delete deep store files only for segments which do not share load
specs with other segments
toolbox.getDataSegmentKiller().kill(segmentsToKillFromDeepStore);
emitMetric(toolbox.getEmitter(),
TaskMetrics.SEGMENTS_DELETED_FROM_DEEPSTORE,
segmentsToKillFromDeepStore.size());
numBatchesProcessed++;
+ totalSegmentsDeletedFromMetadataStore +=
numSegmentsDeletedFromMetadataStore;
numSegmentsKilled += segmentsToKillFromDeepStore.size();
logInfo("Processed [%d] batches for kill task[%s].",
numBatchesProcessed, getId());
@@ -312,8 +329,10 @@ public class KillUnusedSegmentsTask extends
AbstractFixedIntervalTask
final String taskId = getId();
logInfo(
"Finished kill task[%s] for dataSource[%s] and interval[%s]."
- + " Deleted total [%d] unused segments in [%d] batches.",
- taskId, getDataSource(), getInterval(), numSegmentsKilled,
numBatchesProcessed
+ + " Deleted [%d] unused segments from metadata store and files for"
+ + " [%d] segments from deep store in [%d] batches.",
+ taskId, getDataSource(), getInterval(),
+ totalSegmentsDeletedFromMetadataStore, numSegmentsKilled,
numBatchesProcessed
);
final KillTaskReport.Stats stats =
@@ -512,13 +531,13 @@ public class KillUnusedSegmentsTask extends
AbstractFixedIntervalTask
public boolean isReady(TaskActionClient taskActionClient) throws Exception
{
final boolean useConcurrentLocks = Boolean.TRUE.equals(
- getContextValue(
+ QueryContexts.getAsBoolean(
Tasks.USE_CONCURRENT_LOCKS,
- Tasks.DEFAULT_USE_CONCURRENT_LOCKS
+ getContextValue(Tasks.USE_CONCURRENT_LOCKS)
)
);
- TaskLockType actualLockType = determineLockType(useConcurrentLocks);
+ final TaskLockType actualLockType = determineLockType(useConcurrentLocks);
final TaskLock lock = taskActionClient.submit(
new TimeChunkLockTryAcquireAction(
@@ -539,9 +558,13 @@ public class KillUnusedSegmentsTask extends
AbstractFixedIntervalTask
if (useConcurrentLocks) {
actualLockType = TaskLockType.REPLACE;
} else {
- actualLockType = getContextValue(Tasks.TASK_LOCK_TYPE,
TaskLockType.EXCLUSIVE);
+ actualLockType = QueryContexts.getAsEnum(
+ Tasks.TASK_LOCK_TYPE,
+ getContextValue(Tasks.TASK_LOCK_TYPE),
+ TaskLockType.class,
+ TaskLockType.EXCLUSIVE
+ );
}
return actualLockType;
}
-
}
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/duty/UnusedSegmentsKiller.java
b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/duty/UnusedSegmentsKiller.java
index dd23add68d7..a5f8ffa9dc9 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/duty/UnusedSegmentsKiller.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/duty/UnusedSegmentsKiller.java
@@ -34,6 +34,7 @@ import org.apache.druid.indexing.common.task.TaskMetrics;
import org.apache.druid.indexing.common.task.Tasks;
import org.apache.druid.indexing.overlord.GlobalTaskLockbox;
import org.apache.druid.indexing.overlord.IndexerMetadataStorageCoordinator;
+import org.apache.druid.indexing.overlord.config.DefaultTaskConfig;
import org.apache.druid.java.util.common.DateTimes;
import org.apache.druid.java.util.common.Stopwatch;
import org.apache.druid.java.util.common.concurrent.ScheduledExecutorFactory;
@@ -46,15 +47,14 @@ import org.apache.druid.metadata.UnusedSegmentKillerConfig;
import org.apache.druid.query.DruidMetrics;
import org.apache.druid.segment.loading.DataSegmentKiller;
import org.apache.druid.server.http.DataSegmentPlus;
+import org.apache.druid.timeline.DatasourceInterval;
import org.joda.time.DateTime;
import org.joda.time.Duration;
import org.joda.time.Interval;
-import javax.annotation.Nullable;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.Set;
import java.util.concurrent.Future;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
@@ -63,11 +63,12 @@ import java.util.concurrent.atomic.AtomicReference;
/**
* {@link OverlordDuty} to delete unused segments from metadata store and the
* deep storage. Launches {@link EmbeddedKillTask}s to clean unused segments
- * of a single datasource-interval.
+ * of a single datasource-interval. These tasks use EXCLUSIVE locks by default.
*
- * @see SegmentsMetadataManagerConfig to enable the cleanup
- * @see org.apache.druid.server.coordinator.duty.KillUnusedSegments for legacy
- * mode of killing unused segments via Coordinator duties
+ * @see SegmentsMetadataManagerConfig#getKillUnused() Config to enable the
cleanup
+ * @see org.apache.druid.server.coordinator.duty.KillUnusedSegments
+ * Legacy mode of killing segments via Coordinator duties
+ * @see KillUnusedSegmentsTask Behaviour of kill task with concurrent locks
*/
public class UnusedSegmentsKiller implements OverlordDuty
{
@@ -76,8 +77,14 @@ public class UnusedSegmentsKiller implements OverlordDuty
private static final String TASK_ID_PREFIX = "overlord-issued";
private static final int INITIAL_KILL_QUEUE_SIZE = 1000;
- private static final int MAX_INTERVALS_TO_KILL_IN_DATASOURCE = 10_000;
- private static final int MAX_SEGMENTS_TO_KILL_IN_INTERVAL = 1000;
+ private static final int MAX_INTERVALS_TO_KILL = 10_000;
+ private static final int MAX_SEGMENTS_TO_KILL_IN_BATCH = 1000;
+
+ /**
+ * Keep max segments to kill in a single kill task small so that the
EXCLUSIVE
+ * lock on the underlying interval is not held for too long.
+ */
+ private static final int MAX_SEGMENTS_TO_KILL_IN_TASK = 10 *
MAX_SEGMENTS_TO_KILL_IN_BATCH;
/**
* Period after which the queue is reset even if there are existing jobs in
queue.
@@ -85,9 +92,11 @@ public class UnusedSegmentsKiller implements OverlordDuty
private static final Duration QUEUE_RESET_PERIOD = Duration.standardDays(1);
/**
- * Duration for which a kill task is allowed to run.
+ * Duration for which a kill task is allowed to run. Assuming that processing
+ * each batch of segments takes around 30s, each kill task (upto 10 batches)
+ * should normally finish in 5 minutes.
*/
- private static final Duration MAX_TASK_DURATION =
Duration.standardMinutes(10);
+ private static final Duration MAX_TASK_DURATION =
Duration.standardMinutes(30);
private final ServiceEmitter emitter;
private final GlobalTaskLockbox taskLockbox;
@@ -116,6 +125,7 @@ public class UnusedSegmentsKiller implements OverlordDuty
@Inject
public UnusedSegmentsKiller(
SegmentsMetadataManagerConfig config,
+ DefaultTaskConfig defaultTaskConfig,
TaskActionClientFactory taskActionClientFactory,
IndexerMetadataStorageCoordinator storageCoordinator,
@IndexingService DruidLeaderSelector leaderSelector,
@@ -139,7 +149,7 @@ public class UnusedSegmentsKiller implements OverlordDuty
this.killQueue = new PriorityBlockingQueue<>(
INITIAL_KILL_QUEUE_SIZE,
Ordering.from(Comparators.intervalsByEndThenStart())
- .onResultOf(candidate -> candidate.interval)
+ .onResultOf(KillCandidate::interval)
);
} else {
this.exec = null;
@@ -193,7 +203,7 @@ public class UnusedSegmentsKiller implements OverlordDuty
// Schedule the first run after some delay since the segment metadata
cache
// might take time for the sync to finish if this Overlord has just
started.
final long periodMillis =
killConfig.getDutyPeriod().toStandardDuration().getMillis();
- final long initialDelayMillis = periodMillis / 4;
+ final long initialDelayMillis = periodMillis / 2;
log.info(
"Unused segment killer is enabled and will start after [%d] millis."
@@ -245,22 +255,53 @@ public class UnusedSegmentsKiller implements OverlordDuty
return;
}
- final Set<String> dataSources =
storageCoordinator.retrieveAllDatasourceNames();
-
final Map<String, Integer> dataSourceToIntervalCounts = new HashMap<>();
- for (String dataSource : dataSources) {
- storageCoordinator.retrieveSomeUnusedSegmentIntervals(dataSource,
MAX_INTERVALS_TO_KILL_IN_DATASOURCE).forEach(
- interval -> {
- dataSourceToIntervalCounts.merge(dataSource, 1, Integer::sum);
- killQueue.offer(new KillCandidate(dataSource, interval));
- }
+
+ // Identify intervals with unused segments which are eligible for kill
+ final Map<DatasourceInterval, Integer> eligibleIntervals =
+ storageCoordinator.retrieveSomeUnusedSegmentIntervals(
+ DateTimes.nowUtc().minus(killConfig.getBufferPeriod()),
+ MAX_INTERVALS_TO_KILL,
+ killConfig.getMaxSegmentsToKill()
+ );
+
+ // Add kill candidates to the queue
+ eligibleIntervals.forEach((entry, numEligibleSegments) -> {
+ dataSourceToIntervalCounts.merge(entry.dataSource(), 1, Integer::sum);
+
+ // Queue multiple candidates for the same datasource-interval if the
+ // number of segments to kill is large to avoid holding a lock for too
long
+ int remainingSegmentsToKill = numEligibleSegments;
+ while (remainingSegmentsToKill > 0) {
+ int numSegmentsToKill
+ = Math.min(remainingSegmentsToKill,
MAX_SEGMENTS_TO_KILL_IN_TASK);
+ remainingSegmentsToKill -= numSegmentsToKill;
+
+ // If only a few segments remain, add them to the same candidate
+ if (remainingSegmentsToKill < MAX_SEGMENTS_TO_KILL_IN_BATCH) {
+ numSegmentsToKill += remainingSegmentsToKill;
+ remainingSegmentsToKill = 0;
+ }
+
+ killQueue.offer(
+ new KillCandidate(entry.dataSource(), entry.interval(),
numSegmentsToKill)
+ );
+ }
+
+ emitMetric(
+ Metric.ELIGIBLE_UNUSED_SEGMENTS,
+ numEligibleSegments,
+ Map.of(
+ DruidMetrics.DATASOURCE, entry.dataSource(),
+ DruidMetrics.INTERVAL, entry.interval().toString()
+ )
);
- }
+ });
lastResetTime.set(DateTimes.nowUtc());
log.info(
"Queued [%d] kill jobs for [%d] datasources in [%d] millis.",
- killQueue.size(), dataSources.size(), resetDuration.millisElapsed()
+ killQueue.size(), dataSourceToIntervalCounts.size(),
resetDuration.millisElapsed()
);
dataSourceToIntervalCounts.forEach(
(dataSource, intervalCount) -> emitMetric(
@@ -306,8 +347,8 @@ public class UnusedSegmentsKiller implements OverlordDuty
final String taskId = IdUtils.newTaskId(
TASK_ID_PREFIX,
KillUnusedSegmentsTask.TYPE,
- candidate.dataSource,
- candidate.interval
+ candidate.dataSource(),
+ candidate.interval()
);
final Future<?> taskFuture = exec.submit(() -> {
@@ -339,6 +380,7 @@ public class UnusedSegmentsKiller implements OverlordDuty
final ServiceMetricEvent.Builder metricBuilder = new
ServiceMetricEvent.Builder();
IndexTaskUtils.setTaskDimensions(metricBuilder, killTask);
+ metricBuilder.setDimension(DruidMetrics.INTERVAL, candidate.interval());
try {
taskLockbox.add(killTask);
@@ -361,7 +403,7 @@ public class UnusedSegmentsKiller implements OverlordDuty
}
finally {
cleanupLocksSilently(killTask);
- emitMetric(Metric.PROCESSED_KILL_JOBS, 1L,
Map.of(DruidMetrics.DATASOURCE, candidate.dataSource));
+ emitMetric(Metric.PROCESSED_KILL_JOBS, 1L,
Map.of(DruidMetrics.DATASOURCE, candidate.dataSource()));
}
}
@@ -387,16 +429,9 @@ public class UnusedSegmentsKiller implements OverlordDuty
/**
* Represents a single candidate interval that contains unused segments.
*/
- private static class KillCandidate
+ private record KillCandidate(String dataSource, Interval interval, int
numSegmentsToKill)
{
- private final String dataSource;
- private final Interval interval;
- private KillCandidate(String dataSource, Interval interval)
- {
- this.dataSource = dataSource;
- this.interval = interval;
- }
}
/**
@@ -436,33 +471,24 @@ public class UnusedSegmentsKiller implements OverlordDuty
{
super(
taskId,
- candidate.dataSource,
- candidate.interval,
+ candidate.dataSource(),
+ candidate.interval(),
null,
Map.of(Tasks.PRIORITY_KEY,
Tasks.DEFAULT_EMBEDDED_KILL_TASK_PRIORITY),
- null,
- null,
+ MAX_SEGMENTS_TO_KILL_IN_BATCH,
+ candidate.numSegmentsToKill(),
maxUpdatedTimeOfEligibleSegment
);
}
- @Nullable
- @Override
- protected Integer getNumTotalBatches()
- {
- // Do everything in a single batch so that locks are not held for very
long
- return 1;
- }
-
@Override
protected List<DataSegmentPlus> fetchNextBatchOfUnusedSegments(TaskToolbox
toolbox, int nextBatchSize)
{
- // Kill only 1000 segments in the batch so that locks are not held for
very long
return storageCoordinator.retrieveUnusedSegmentsWithExactInterval(
getDataSource(),
getInterval(),
getMaxUsedStatusLastUpdatedTime(),
- MAX_SEGMENTS_TO_KILL_IN_INTERVAL
+ nextBatchSize
);
}
@@ -503,5 +529,6 @@ public class UnusedSegmentsKiller implements OverlordDuty
public static final String SKIPPED_INTERVALS =
"segment/kill/skippedIntervals/count";
public static final String UNUSED_SEGMENT_INTERVALS =
"segment/kill/unusedIntervals/count";
+ public static final String ELIGIBLE_UNUSED_SEGMENTS =
"segment/kill/eligibleSegments/count";
}
}
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManager.java
b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManager.java
index 7c75e5a9642..d99fcfa9e68 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManager.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManager.java
@@ -32,7 +32,7 @@ import org.apache.druid.error.DruidException;
import org.apache.druid.error.InvalidInput;
import org.apache.druid.error.NotFound;
import org.apache.druid.guice.annotations.Json;
-import org.apache.druid.indexing.common.TaskLockType;
+import org.apache.druid.indexing.common.actions.TaskLocks;
import org.apache.druid.indexing.common.task.Tasks;
import org.apache.druid.indexing.overlord.DataSourceMetadata;
import
org.apache.druid.indexing.overlord.supervisor.autoscaler.SupervisorTaskAutoScaler;
@@ -49,7 +49,6 @@ import org.apache.druid.java.util.emitter.EmittingLogger;
import org.apache.druid.metadata.MetadataSupervisorManager;
import org.apache.druid.metadata.PendingSegmentRecord;
import org.apache.druid.query.DefaultQueryMetrics;
-import org.apache.druid.query.QueryContexts;
import org.apache.druid.segment.incremental.ParseExceptionReport;
import org.apache.druid.server.metrics.SupervisorStatsProvider;
@@ -778,22 +777,9 @@ public class SupervisorManager implements
SupervisorStatsProvider
*/
private static boolean specHasConcurrentLocks(SeekableStreamSupervisorSpec
spec)
{
- Map<String, Object> context = spec.getContext();
- if (context == null) {
- return Tasks.DEFAULT_USE_CONCURRENT_LOCKS;
- }
- Boolean useConcurrentLocks = QueryContexts.getAsBoolean(
- Tasks.USE_CONCURRENT_LOCKS,
- context.get(Tasks.USE_CONCURRENT_LOCKS)
- );
- if (useConcurrentLocks != null) {
- return useConcurrentLocks;
- }
- TaskLockType taskLockType = QueryContexts.getAsEnum(
- Tasks.TASK_LOCK_TYPE,
- context.get(Tasks.TASK_LOCK_TYPE),
- TaskLockType.class
+ return TaskLocks.shouldUseConcurrentLocksForAppend(
+ spec.getContext(),
+ Tasks.DEFAULT_USE_CONCURRENT_LOCKS
);
- return taskLockType == TaskLockType.APPEND;
}
}
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/KillUnusedSegmentsTaskTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/KillUnusedSegmentsTaskTest.java
index a84d28d2f15..90070060b9c 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/KillUnusedSegmentsTaskTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/KillUnusedSegmentsTaskTest.java
@@ -1406,7 +1406,7 @@ public class KillUnusedSegmentsTaskTest extends
IngestionTestBase
{
final KillUnusedSegmentsTask task = new KillUnusedSegmentsTaskBuilder()
.dataSource(DATA_SOURCE)
- .context(ImmutableMap.of(Tasks.TASK_LOCK_TYPE, TaskLockType.APPEND))
+ .context(ImmutableMap.of(Tasks.TASK_LOCK_TYPE,
TaskLockType.APPEND.name()))
.interval(Intervals.of("2019-03-01/2019-04-01"))
.build();
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/duty/UnusedSegmentsKillerTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/duty/UnusedSegmentsKillerTest.java
index 7f0d718b0b9..1c8d885c924 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/duty/UnusedSegmentsKillerTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/duty/UnusedSegmentsKillerTest.java
@@ -28,6 +28,7 @@ import org.apache.druid.indexing.common.task.TaskMetrics;
import org.apache.druid.indexing.overlord.GlobalTaskLockbox;
import org.apache.druid.indexing.overlord.IndexerMetadataStorageCoordinator;
import org.apache.druid.indexing.overlord.TimeChunkLockRequest;
+import org.apache.druid.indexing.overlord.config.DefaultTaskConfig;
import org.apache.druid.indexing.test.TestDataSegmentKiller;
import org.apache.druid.java.util.common.ISE;
import org.apache.druid.java.util.common.Intervals;
@@ -82,7 +83,7 @@ public class UnusedSegmentsKillerTest
emitter = taskActionTestKit.getServiceEmitter();
leaderSelector = new TestDruidLeaderSelector();
dataSegmentKiller = new TestDataSegmentKiller();
- killerConfig = new UnusedSegmentKillerConfig(true, Period.ZERO, null);
+ killerConfig = new UnusedSegmentKillerConfig(true, Period.ZERO, null,
null);
killExecutor = new BlockingExecutorService("UnusedSegmentsKillerTest-%s");
storageCoordinator = taskActionTestKit.getMetadataStorageCoordinator();
initKiller();
@@ -96,6 +97,7 @@ public class UnusedSegmentsKillerTest
SegmentMetadataCache.UsageMode.ALWAYS,
killerConfig
),
+ new DefaultTaskConfig(),
taskActionTestKit::createTaskActionClient,
storageCoordinator,
leaderSelector,
@@ -106,6 +108,14 @@ public class UnusedSegmentsKillerTest
);
}
+ private void resetKillQueue()
+ {
+ killer.run();
+
+ // Invoke rebuild of kill queue on the executor thread
+ killExecutor.finishNextPendingTask();
+ }
+
private void finishQueuedKillJobs()
{
killExecutor.finishAllPendingTasks();
@@ -116,13 +126,13 @@ public class UnusedSegmentsKillerTest
{
final DutySchedule schedule = killer.getSchedule();
Assert.assertEquals(Duration.standardHours(1).getMillis(),
schedule.getPeriodMillis());
- Assert.assertEquals(Duration.standardMinutes(15).getMillis(),
schedule.getInitialDelayMillis());
+ Assert.assertEquals(Duration.standardMinutes(30).getMillis(),
schedule.getInitialDelayMillis());
}
@Test
public void test_getSchedule_returnsZeroPeriod_ifDisabled()
{
- killerConfig = new UnusedSegmentKillerConfig(false, null, null);
+ killerConfig = new UnusedSegmentKillerConfig(false, null, null, null);
initKiller();
final DutySchedule schedule = killer.getSchedule();
@@ -143,7 +153,7 @@ public class UnusedSegmentsKillerTest
@Test
public void test_run_isNoop_ifDisabled()
{
- killerConfig = new UnusedSegmentKillerConfig(false, null, null);
+ killerConfig = new UnusedSegmentKillerConfig(false, null, null, null);
initKiller();
Assert.assertFalse(killer.isEnabled());
@@ -201,68 +211,96 @@ public class UnusedSegmentsKillerTest
}
@Test
- public void test_maxSegmentsKilledInAnInterval_is_1k()
+ public void test_maxSegmentsKilledInRun_isLimitedByConfig()
{
+ killerConfig = new UnusedSegmentKillerConfig(true, Period.ZERO, null, 700);
+ initKiller();
leaderSelector.becomeLeader();
final List<DataSegment> segments =
CreateDataSegments.ofDatasource(TestDataSource.WIKI)
- .forIntervals(1, Granularities.DAY)
- .withNumPartitions(2000)
+ .forIntervals(10, Granularities.DAY)
+ .withNumPartitions(100)
.eachOfSizeInMb(50);
storageCoordinator.commitSegments(Set.copyOf(segments), null);
storageCoordinator.markAllSegmentsAsUnused(TestDataSource.WIKI);
Assert.assertEquals(
- 2000,
- retrieveUnusedSegments(segments.get(0).getInterval()).size()
+ 1000,
+ retrieveUnusedSegments(Intervals.ETERNITY).size()
);
- // Reset the kill queue and execute kill tasks
- killer.run();
+ resetKillQueue();
finishQueuedKillJobs();
- // Verify that a single kill task has run which killed 1k segments
- emitter.verifyEmitted(TaskMetrics.RUN_DURATION, 1);
- emitter.verifySum(TaskMetrics.SEGMENTS_DELETED_FROM_METADATA_STORE, 1000L);
-
- Assert.assertEquals(
- 1000,
- retrieveUnusedSegments(segments.get(0).getInterval()).size()
- );
+ // Verify that a total of 700 segments were identified for kill
+ emitter.verifySum(UnusedSegmentsKiller.Metric.ELIGIBLE_UNUSED_SEGMENTS,
700L);
+ emitter.verifySum(TaskMetrics.SEGMENTS_DELETED_FROM_DEEPSTORE, 700L);
}
@Test(timeout = 20_000L)
- public void test_maxIntervalsKilledInADatasource_is_10k()
+ public void test_maxSegmentsKilledByTask_is_10k()
{
leaderSelector.becomeLeader();
+ // Create an interval with 12k killable unused segments
final List<DataSegment> segments =
CreateDataSegments.ofDatasource(TestDataSource.WIKI)
- .forIntervals(20_000, Granularities.DAY)
+ .forIntervals(1, Granularities.DAY)
+ .withNumPartitions(12_000)
.eachOfSizeInMb(50);
storageCoordinator.commitSegments(Set.copyOf(segments), null);
storageCoordinator.markAllSegmentsAsUnused(TestDataSource.WIKI);
Assert.assertEquals(
- 20_000,
+ 12_000,
retrieveUnusedSegments(Intervals.ETERNITY).size()
);
- // Reset the kill queue and execute kill tasks
- killer.run();
+ resetKillQueue();
finishQueuedKillJobs();
- // Verify that 10k kill tasks have run, each killing a single segment
- emitter.verifyEmitted(TaskMetrics.RUN_DURATION, 10000);
- emitter.verifySum(TaskMetrics.SEGMENTS_DELETED_FROM_METADATA_STORE,
10_000L);
+ // Verify that all the segments in the interval were eligible for kill
+ emitter.verifySum(UnusedSegmentsKiller.Metric.UNUSED_SEGMENT_INTERVALS,
1L);
+ emitter.verifySum(UnusedSegmentsKiller.Metric.ELIGIBLE_UNUSED_SEGMENTS,
12_000L);
+
+ // Verify that 2 tasks were launched to kill the segments
+ emitter.verifySum(UnusedSegmentsKiller.Metric.PROCESSED_KILL_JOBS, 2L);
+ final List<String> taskIds =
emitter.getMetricEvents(TaskMetrics.RUN_DURATION)
+ .stream()
+ .map(event ->
event.getUserDims().get("taskId").toString())
+ .toList();
+ Assert.assertEquals(2, taskIds.size());
+
+ // Verify that the tasks killed 10k and 2k segments respectively
+ emitter.verifySum(TaskMetrics.SEGMENTS_DELETED_FROM_DEEPSTORE,
Map.of("taskId", taskIds.get(0)), 10_000);
+ emitter.verifySum(TaskMetrics.SEGMENTS_DELETED_FROM_DEEPSTORE,
Map.of("taskId", taskIds.get(1)), 2_000);
+ }
+
+ @Test(timeout = 20_000L)
+ public void test_maxIntervalsKilledInADatasource_is_10k()
+ {
+ leaderSelector.becomeLeader();
+
+ final List<DataSegment> segments =
+ CreateDataSegments.ofDatasource(TestDataSource.WIKI)
+ .forIntervals(10_001, Granularities.DAY)
+ .eachOfSizeInMb(50);
+
+ storageCoordinator.commitSegments(Set.copyOf(segments), null);
+ storageCoordinator.markAllSegmentsAsUnused(TestDataSource.WIKI);
Assert.assertEquals(
- 10_000,
+ 10_001,
retrieveUnusedSegments(Intervals.ETERNITY).size()
);
+
+ resetKillQueue();
+
+ emitter.verifySum(UnusedSegmentsKiller.Metric.UNUSED_SEGMENT_INTERVALS,
10_000L);
+ emitter.verifySum(UnusedSegmentsKiller.Metric.ELIGIBLE_UNUSED_SEGMENTS,
10_000L);
}
@Test
@@ -398,7 +436,7 @@ public class UnusedSegmentsKillerTest
@Test
public void test_run_doesNotKillSegment_ifUpdatedWithinBufferPeriod()
{
- killerConfig = new UnusedSegmentKillerConfig(true, Period.hours(1), null);
+ killerConfig = new UnusedSegmentKillerConfig(true, Period.hours(1), null,
null);
initKiller();
storageCoordinator.commitSegments(Set.copyOf(WIKI_SEGMENTS_1X10D), null);
@@ -408,13 +446,12 @@ public class UnusedSegmentsKillerTest
killer.run();
finishQueuedKillJobs();
- // Verify that tasks are launched but no segment is killed
- emitter.verifyValue(UnusedSegmentsKiller.Metric.UNUSED_SEGMENT_INTERVALS,
10L);
- emitter.verifyEmitted(UnusedSegmentsKiller.Metric.PROCESSED_KILL_JOBS, 10);
- emitter.verifyEmitted(TaskMetrics.RUN_DURATION, 10);
+ // Verify that no tasks are launched
+
emitter.verifyNotEmitted(UnusedSegmentsKiller.Metric.UNUSED_SEGMENT_INTERVALS);
+
emitter.verifyNotEmitted(UnusedSegmentsKiller.Metric.ELIGIBLE_UNUSED_SEGMENTS);
- emitter.verifySum(TaskMetrics.SEGMENTS_DELETED_FROM_METADATA_STORE, 0L);
- emitter.verifySum(TaskMetrics.SEGMENTS_DELETED_FROM_DEEPSTORE, 0L);
+ emitter.verifyNotEmitted(TaskMetrics.SEGMENTS_DELETED_FROM_METADATA_STORE);
+ emitter.verifyNotEmitted(TaskMetrics.SEGMENTS_DELETED_FROM_DEEPSTORE);
}
@Test
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/http/OverlordDataSourcesResourceTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/http/OverlordDataSourcesResourceTest.java
index 53af56961d2..5070197c37a 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/http/OverlordDataSourcesResourceTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/http/OverlordDataSourcesResourceTest.java
@@ -43,7 +43,6 @@ import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Set;
-import java.util.stream.Collectors;
public class OverlordDataSourcesResourceTest
{
@@ -210,7 +209,7 @@ public class OverlordDataSourcesResourceTest
final String version2 = WIKI_SEGMENTS_10X1D.get(0).getVersion() + "__2";
final List<DataSegment> wikiSegmentsV2 = WIKI_SEGMENTS_10X1D.stream().map(
segment -> DataSegment.builder(segment).version(version2).build()
- ).collect(Collectors.toList());
+ ).toList();
storageCoordinator.commitSegments(Set.copyOf(wikiSegmentsV2), null);
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/test/TestIndexerMetadataStorageCoordinator.java
b/indexing-service/src/test/java/org/apache/druid/indexing/test/TestIndexerMetadataStorageCoordinator.java
index 41a30024163..803be668960 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/test/TestIndexerMetadataStorageCoordinator.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/test/TestIndexerMetadataStorageCoordinator.java
@@ -38,6 +38,7 @@ import
org.apache.druid.segment.realtime.appenderator.SegmentIdWithShardSpec;
import
org.apache.druid.server.coordinator.simulate.TestSegmentsMetadataManager;
import org.apache.druid.server.http.DataSegmentPlus;
import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.timeline.DatasourceInterval;
import org.apache.druid.timeline.SegmentId;
import org.apache.druid.timeline.SegmentTimeline;
import org.joda.time.DateTime;
@@ -75,6 +76,16 @@ public class TestIndexerMetadataStorageCoordinator
implements IndexerMetadataSto
return List.of();
}
+ @Override
+ public Map<DatasourceInterval, Integer> retrieveSomeUnusedSegmentIntervals(
+ DateTime maxUpdatedTime,
+ int maxResultSize,
+ int maxSegmentsToScan
+ )
+ {
+ return Map.of();
+ }
+
@Override
public List<DataSegmentPlus> retrieveUnusedSegmentsWithExactInterval(
String dataSource,
diff --git
a/processing/src/main/java/org/apache/druid/timeline/DatasourceInterval.java
b/processing/src/main/java/org/apache/druid/timeline/DatasourceInterval.java
new file mode 100644
index 00000000000..0e7796c1646
--- /dev/null
+++ b/processing/src/main/java/org/apache/druid/timeline/DatasourceInterval.java
@@ -0,0 +1,29 @@
+/*
+ * 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.timeline;
+
+import org.joda.time.Interval;
+
+/**
+ * A tuple containing a datasource and an interval.
+ */
+public record DatasourceInterval(String dataSource, Interval interval)
+{
+}
diff --git
a/server/src/main/java/org/apache/druid/indexing/overlord/IndexerMetadataStorageCoordinator.java
b/server/src/main/java/org/apache/druid/indexing/overlord/IndexerMetadataStorageCoordinator.java
index 26dcae7cd1f..2d0193796e6 100644
---
a/server/src/main/java/org/apache/druid/indexing/overlord/IndexerMetadataStorageCoordinator.java
+++
b/server/src/main/java/org/apache/druid/indexing/overlord/IndexerMetadataStorageCoordinator.java
@@ -28,6 +28,7 @@ import org.apache.druid.segment.SegmentSchemaMapping;
import org.apache.druid.segment.realtime.appenderator.SegmentIdWithShardSpec;
import org.apache.druid.server.http.DataSegmentPlus;
import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.timeline.DatasourceInterval;
import org.apache.druid.timeline.SegmentId;
import org.apache.druid.timeline.SegmentTimeline;
import org.joda.time.DateTime;
@@ -618,6 +619,16 @@ public interface IndexerMetadataStorageCoordinator
*/
List<Interval> retrieveSomeUnusedSegmentIntervals(String dataSource, int
limit);
+ /**
+ * Extracts eligible unused segments intervals from the metadata store using
+ * {@link
org.apache.druid.metadata.SqlSegmentsMetadataQuery#retrieveSomeUnusedSegmentIntervals(DateTime,
int, int)}
+ */
+ Map<DatasourceInterval, Integer> retrieveSomeUnusedSegmentIntervals(
+ DateTime maxUpdatedTime,
+ int maxResultSize,
+ int maxSegmentsToScan
+ );
+
/**
* Returns the number of segment entries in the database whose state was
changed as the result of this call (that is,
* the segments were marked as used). If the call results in a database
error, an exception is relayed to the caller.
diff --git
a/server/src/main/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinator.java
b/server/src/main/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinator.java
index fe04a9a0c1d..49e18862c05 100644
---
a/server/src/main/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinator.java
+++
b/server/src/main/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinator.java
@@ -61,6 +61,7 @@ import org.apache.druid.segment.metadata.SegmentSchemaManager;
import org.apache.druid.segment.realtime.appenderator.SegmentIdWithShardSpec;
import org.apache.druid.server.http.DataSegmentPlus;
import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.timeline.DatasourceInterval;
import org.apache.druid.timeline.Partitions;
import org.apache.druid.timeline.SegmentId;
import org.apache.druid.timeline.SegmentTimeline;
@@ -171,6 +172,18 @@ public class IndexerSQLMetadataStorageCoordinator
implements IndexerMetadataStor
);
}
+ @Override
+ public Map<DatasourceInterval, Integer> retrieveSomeUnusedSegmentIntervals(
+ DateTime maxUpdatedTime,
+ int maxResultSize,
+ int maxSegmentsToScan
+ )
+ {
+ return inReadOnlyTransaction(
+ sql -> sql.retrieveSomeUnusedSegmentIntervals(maxUpdatedTime,
maxResultSize, maxSegmentsToScan)
+ );
+ }
+
@Override
public Set<DataSegment> retrieveUsedSegmentsForIntervals(
final String dataSource,
diff --git
a/server/src/main/java/org/apache/druid/metadata/SegmentsMetadataManagerConfig.java
b/server/src/main/java/org/apache/druid/metadata/SegmentsMetadataManagerConfig.java
index 8fd23da5e15..d02bd6bdec2 100644
---
a/server/src/main/java/org/apache/druid/metadata/SegmentsMetadataManagerConfig.java
+++
b/server/src/main/java/org/apache/druid/metadata/SegmentsMetadataManagerConfig.java
@@ -52,7 +52,7 @@ public class SegmentsMetadataManagerConfig
{
this.pollDuration = Configs.valueOrDefault(pollDuration,
Period.minutes(1));
this.useIncrementalCache = Configs.valueOrDefault(useIncrementalCache,
SegmentMetadataCache.UsageMode.IF_SYNCED);
- this.killUnused = Configs.valueOrDefault(killUnused, new
UnusedSegmentKillerConfig(null, null, null));
+ this.killUnused = Configs.valueOrDefault(killUnused, new
UnusedSegmentKillerConfig(null, null, null, null));
if (this.killUnused.isEnabled() && this.useIncrementalCache ==
SegmentMetadataCache.UsageMode.NEVER) {
throw DruidException
.forPersona(DruidException.Persona.OPERATOR)
diff --git
a/server/src/main/java/org/apache/druid/metadata/SqlSegmentsMetadataQuery.java
b/server/src/main/java/org/apache/druid/metadata/SqlSegmentsMetadataQuery.java
index dfeb376204d..d16ba18d0b2 100644
---
a/server/src/main/java/org/apache/druid/metadata/SqlSegmentsMetadataQuery.java
+++
b/server/src/main/java/org/apache/druid/metadata/SqlSegmentsMetadataQuery.java
@@ -36,6 +36,7 @@ import org.apache.druid.java.util.common.DateTimes;
import org.apache.druid.java.util.common.IAE;
import org.apache.druid.java.util.common.Intervals;
import org.apache.druid.java.util.common.JodaUtils;
+import org.apache.druid.java.util.common.Pair;
import org.apache.druid.java.util.common.StringUtils;
import org.apache.druid.java.util.common.jackson.JacksonUtils;
import org.apache.druid.java.util.common.logger.Logger;
@@ -48,6 +49,7 @@ import
org.apache.druid.segment.realtime.appenderator.SegmentIdWithShardSpec;
import org.apache.druid.server.http.DataSegmentPlus;
import org.apache.druid.timeline.CompactionState;
import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.timeline.DatasourceInterval;
import org.apache.druid.timeline.SegmentId;
import org.apache.druid.timeline.SegmentTimeline;
import org.apache.druid.utils.CloseableUtils;
@@ -1079,7 +1081,7 @@ public class SqlSegmentsMetadataQuery
*
* @return List of distinct unused segment intervals for the specified
datasource
* containing at least 1 entry if there is any unused segment for the
datasource,
- * upto a maximum of {@code limit} entries.
+ * up to a maximum of {@code limit} entries.
*/
public List<Interval> retrieveSomeUnusedSegmentIntervals(String dataSource,
int limit)
{
@@ -1116,6 +1118,70 @@ public class SqlSegmentsMetadataQuery
return
intervals.stream().filter(Objects::nonNull).collect(Collectors.toList());
}
+ /**
+ * Scans up to {@code maxSegmentsToScan} unused segments which are eligible
for
+ * kill and returns the unique datasource-interval for the segments scanned.
+ * <p>
+ * This method ensures that if there is any unused segment in any datasource
+ * which was updated earlier than {@code maxUpdatedTime}, then the returned
+ * map is not empty. However, it does NOT guarantee that:
+ * <ul>
+ * <li>the candidates in the returned map would be ordered by datasource or
interval</li>
+ * <li>the result would contain {@code maxResultSize} entries when there are
more distinct
+ * intervals with eligible unused segments in the metadata store.</li>
+ * </ul>
+ *
+ * @param maxUpdatedTime Unused segments are considered eligible for kill
+ * if they were last updated before this time.
+ * @param maxResultSize Maximum number of candidate intervals to return
+ * across all datasources.
+ * @param maxSegmentsToScan Maximum number of eligible unused segments to
scan
+ * in the metadata store.
+ * @return Map from {@link DatasourceInterval} to the number of unused
segments
+ * eligible for kill.
+ */
+ public Map<DatasourceInterval, Integer> retrieveSomeUnusedSegmentIntervals(
+ DateTime maxUpdatedTime,
+ int maxResultSize,
+ int maxSegmentsToScan
+ )
+ {
+ final String sql = StringUtils.format(
+ // Disable checkstyle to avoid argumentLineBreaking rule from getting
triggered
+ //CHECKSTYLE.OFF: Regexp
+ """
+ SELECT dataSource, start, %2$send%2$s, COUNT(*) AS totalCount
+ FROM (
+ SELECT dataSource, start, %2$send%2$s
+ FROM %1$s
+ WHERE used = false
+ AND (used_status_last_updated IS NULL OR
used_status_last_updated <= :maxUpdatedTime)
+ %3$s
+ ) AS unused
+ GROUP BY dataSource, %2$send%2$s, start
+ %4$s
+ """,
+ //CHECKSTYLE.ON: Regexp
+ dbTables.getSegmentsTable(),
+ connector.getQuoteString(),
+ connector.limitClause(maxSegmentsToScan),
+ connector.limitClause(maxResultSize)
+ );
+
+ final List<Pair<DatasourceInterval, Integer>> unusedSegmentIntervals =
connector.inReadOnlyTransaction(
+ (handle, status) ->
+ handle.createQuery(sql)
+ .setFetchSize(connector.getStreamingFetchSize())
+ .bind("maxUpdatedTime", maxUpdatedTime.toString())
+ .map((index, r, ctx) -> mapToUnusedSegmentInterval(r))
+ .list()
+ );
+
+ return unusedSegmentIntervals.stream().filter(Objects::nonNull).collect(
+ Collectors.toMap(pair -> pair.lhs, pair -> pair.rhs)
+ );
+ }
+
/**
* Retrieves unused segments that exactly match the given interval.
*
@@ -1691,6 +1757,9 @@ public class SqlSegmentsMetadataQuery
return sql;
}
+ /**
+ * Reads the fields {@code start} and {@code end} from the given result set.
+ */
@Nullable
private Interval mapToInterval(ResultSet resultSet, String dataSource)
{
@@ -1706,6 +1775,30 @@ public class SqlSegmentsMetadataQuery
}
}
+ /**
+ * Reads the fields {@code dataSource}, {@code start}, {@code end} and
+ * {@code totalCount} from the given result set.
+ */
+ @Nullable
+ private Pair<DatasourceInterval, Integer>
mapToUnusedSegmentInterval(ResultSet resultSet)
+ {
+ try {
+ final String dataSource = resultSet.getString("dataSource");
+ final int totalCount = resultSet.getInt("totalCount");
+
+ final Interval interval = mapToInterval(resultSet, "");
+ if (interval == null) {
+ return null;
+ } else {
+ return Pair.of(new DatasourceInterval(dataSource, interval),
totalCount);
+ }
+ }
+ catch (Throwable t) {
+ log.error(t, "Could not read unused segment interval record");
+ return null;
+ }
+ }
+
/**
* Tries to parse the fields of the result set into a {@link
SegmentSchemaRecord}.
*
diff --git
a/server/src/main/java/org/apache/druid/metadata/UnusedSegmentKillerConfig.java
b/server/src/main/java/org/apache/druid/metadata/UnusedSegmentKillerConfig.java
index b355795e0b8..b2deed50438 100644
---
a/server/src/main/java/org/apache/druid/metadata/UnusedSegmentKillerConfig.java
+++
b/server/src/main/java/org/apache/druid/metadata/UnusedSegmentKillerConfig.java
@@ -22,6 +22,7 @@ package org.apache.druid.metadata;
import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
import org.apache.druid.common.config.Configs;
+import org.apache.druid.error.InvalidInput;
import org.apache.druid.java.util.common.logger.Logger;
import org.joda.time.Period;
@@ -35,6 +36,14 @@ public class UnusedSegmentKillerConfig
{
private static final Logger log = new
Logger(UnusedSegmentKillerConfig.class);
+ /**
+ * Maximum number of segments that should be killed in a single run of the
duty.
+ * A value of 200k typically causes the search query to finish within ~5s and
+ * allows the kill queue to finish processing in about (200 * 30s) = 100
minutes,
+ * since a batch of 1000 segments takes about 30s to be processed.
+ */
+ public static final int DEFAULT_MAX_SEGMENTS_TO_KILL = 200_000;
+
@JsonProperty("enabled")
private final boolean enabled;
@@ -44,15 +53,33 @@ public class UnusedSegmentKillerConfig
@JsonProperty("dutyPeriod")
private final Period dutyPeriod;
+ @JsonProperty("maxSegmentsToKill")
+ private final Integer maxSegmentsToKill;
+
@JsonCreator
public UnusedSegmentKillerConfig(
@JsonProperty("enabled") @Nullable Boolean enabled,
@JsonProperty("bufferPeriod") @Nullable Period bufferPeriod,
- @JsonProperty("dutyPeriod") @Nullable Period dutyPeriod
+ @JsonProperty("dutyPeriod") @Nullable Period dutyPeriod,
+ @JsonProperty("maxSegmentsToKill") @Nullable Integer maxSegmentsToKill
)
{
this.enabled = Configs.valueOrDefault(enabled, false);
this.bufferPeriod = Configs.valueOrDefault(bufferPeriod, Period.days(30));
+ this.maxSegmentsToKill = Configs.valueOrDefault(maxSegmentsToKill,
DEFAULT_MAX_SEGMENTS_TO_KILL);
+
+ if (this.maxSegmentsToKill > DEFAULT_MAX_SEGMENTS_TO_KILL) {
+ log.warn(
+ "Setting a high value[%d] for
'druid.manager.segments.killUnused.maxSegmentsToKill'."
+ + " This may slow down the segment killer and/or put undue strain on
the metadata store.",
+ this.maxSegmentsToKill
+ );
+ } else {
+ InvalidInput.conditionalException(
+ this.maxSegmentsToKill > 0,
+ "'druid.manager.segments.killUnused.maxSegmentsToKill' must be
greater than zero"
+ );
+ }
if (dutyPeriod == null) {
this.dutyPeriod = Period.hours(1);
@@ -68,6 +95,7 @@ public class UnusedSegmentKillerConfig
/**
* Period for which segments are retained even after being marked as unused.
+ * Default value is 30 days.
*/
public Period getBufferPeriod()
{
@@ -84,6 +112,15 @@ public class UnusedSegmentKillerConfig
return dutyPeriod;
}
+ /**
+ * Maximum number of segments to kill in a single run of the duty. Default
+ * value is {@link #DEFAULT_MAX_SEGMENTS_TO_KILL}.
+ */
+ public int getMaxSegmentsToKill()
+ {
+ return maxSegmentsToKill;
+ }
+
public boolean isEnabled()
{
return enabled;
diff --git
a/server/src/test/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinatorTest.java
b/server/src/test/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinatorTest.java
index 33b65e55f80..45f02c3713a 100644
---
a/server/src/test/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinatorTest.java
+++
b/server/src/test/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinatorTest.java
@@ -63,6 +63,7 @@ import
org.apache.druid.server.coordinator.simulate.WrappingScheduledExecutorSer
import org.apache.druid.server.http.DataSegmentPlus;
import org.apache.druid.timeline.CompactionState;
import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.timeline.DatasourceInterval;
import org.apache.druid.timeline.SegmentId;
import org.apache.druid.timeline.SegmentTimeline;
import org.apache.druid.timeline.partition.DimensionRangeShardSpec;
@@ -2252,7 +2253,7 @@ public class IndexerSQLMetadataStorageCoordinatorTest
extends IndexerSqlMetadata
}
@Test
- public void testRetrieveSomeUnusedSegmentIntervals()
+ public void testRetrieveSomeUnusedSegmentIntervalsForDatasource()
{
final String dataSource = defaultSegment.getDataSource();
coordinator.commitSegments(Set.of(defaultSegment, defaultSegment3), null);
@@ -2278,6 +2279,38 @@ public class IndexerSQLMetadataStorageCoordinatorTest
extends IndexerSqlMetadata
);
}
+ @Test
+ public void testRetrieveSomeUnusedSegmentIntervals()
+ {
+ final String dataSource = defaultSegment.getDataSource();
+ coordinator.commitSegments(Set.of(defaultSegment, defaultSegment3), null);
+
+ final DateTime maxUpdatedTime = DateTimes.nowUtc();
+
+
Assert.assertTrue(coordinator.retrieveSomeUnusedSegmentIntervals(maxUpdatedTime,
100, 100).isEmpty());
+
+ markAllSegmentsUnused(Set.of(defaultSegment),
DateTimes.nowUtc().minusHours(1));
+ Assert.assertEquals(
+ Map.of(new DatasourceInterval(dataSource,
defaultSegment.getInterval()), 1),
+ coordinator.retrieveSomeUnusedSegmentIntervals(maxUpdatedTime, 100,
100)
+ );
+
+ markAllSegmentsUnused(Set.of(defaultSegment3),
DateTimes.nowUtc().minusHours(1));
+ Assert.assertEquals(
+ Map.of(
+ new DatasourceInterval(dataSource, defaultSegment.getInterval()),
1,
+ new DatasourceInterval(dataSource, defaultSegment3.getInterval()),
1
+ ),
+ coordinator.retrieveSomeUnusedSegmentIntervals(maxUpdatedTime, 100,
100)
+ );
+
+ // Verify retrieve with limit 1 returns only 1 interval
+ Assert.assertEquals(
+ 1,
+ coordinator.retrieveSomeUnusedSegmentIntervals(maxUpdatedTime, 1,
1).size()
+ );
+ }
+
@Test
public void testRetrieveAllDatasourceNames()
{
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
