fix: Stop kill task if segment IDs that share load spec of killable segment could not be determined #19737
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -48,10 +48,10 @@ | |
| import org.apache.druid.java.util.common.StringUtils; | ||
| import org.apache.druid.java.util.common.logger.Logger; | ||
| import org.apache.druid.server.coordination.BroadcastDatasourceLoadingSpec; | ||
| import org.apache.druid.server.http.DataSegmentPlus; | ||
| import org.apache.druid.server.lookup.cache.LookupLoadingSpec; | ||
| import org.apache.druid.server.security.ResourceAction; | ||
| import org.apache.druid.timeline.DataSegment; | ||
| import org.apache.druid.utils.CollectionUtils; | ||
| import org.joda.time.DateTime; | ||
| import org.joda.time.Interval; | ||
|
|
@@ -67,10 +67,10 @@ | |
| import java.util.NavigableMap; | ||
| import java.util.Set; | ||
| import java.util.TreeMap; | ||
| import java.util.function.Function; | ||
| import java.util.stream.Collectors; | ||
|
|
||
| /** | ||
| * The client representation of this task is {@link ClientKillUnusedSegmentsTaskQuery}. | ||
| * JSON serialization fields of this class must correspond to those of {@link | ||
| * ClientKillUnusedSegmentsTaskQuery}, except for {@link #id} and {@link #context} fields. | ||
|
|
@@ -87,6 +87,10 @@ | |
| * <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. | ||
| */ | ||
| public class KillUnusedSegmentsTask extends AbstractFixedIntervalTask | ||
| { | ||
|
|
@@ -211,7 +215,7 @@ public TaskStatus runTask(TaskToolbox toolbox) throws Exception | |
| // List unused segments | ||
| int nextBatchSize = computeNextBatchSize(numSegmentsKilled); | ||
| @Nullable Integer numTotalBatches = getNumTotalBatches(); | ||
| List<DataSegmentPlus> unusedSegmentsPlus; | ||
| logInfo( | ||
| "Starting kill for datasource[%s] in interval[%s] and versions[%s] with batchSize[%d], up to limit[%d]" | ||
| + " segments before maxUsedStatusLastUpdatedTime[%s] will be deleted%s", | ||
|
|
@@ -236,12 +240,25 @@ public TaskStatus runTask(TaskToolbox toolbox) throws Exception | |
| break; | ||
| } | ||
|
|
||
| unusedSegmentsPlus = fetchNextBatchOfUnusedSegments(toolbox, nextBatchSize); | ||
| if (unusedSegmentsPlus.isEmpty()) { | ||
| // No more segments eligible for kill, do not proceed further | ||
| break; | ||
| } | ||
|
|
||
| // Fetch locks each time as a revokal could have occurred in between batches | ||
| final NavigableMap<DateTime, List<TaskLock>> taskLockMap | ||
| = getNonRevokedTaskLockMap(toolbox.getTaskActionClient()); | ||
|
|
||
| final Set<DataSegment> unusedSegments = unusedSegmentsPlus.stream() | ||
| .map(DataSegmentPlus::getDataSegment) | ||
| .collect(Collectors.toSet()); | ||
| final Map<String, DataSegmentPlus> unusedIdToSegmentPlus = CollectionUtils.toMap( | ||
| unusedSegmentsPlus, | ||
| segment -> segment.getDataSegment().getId().toString(), | ||
| Function.identity() | ||
| ); | ||
|
|
||
| if (!TaskLocks.isLockCoversSegments(taskLockMap, unusedSegments)) { | ||
|
Comment on lines
+253
to
262
Contributor
There was a problem hiding this comment. nit: perhaps we could introduce a thin wrapper for
Contributor
Author
There was a problem hiding this comment. The |
||
| throw new ISE( | ||
| "Locks[%s] for task[%s] can't cover segments[%s]", | ||
|
|
@@ -251,62 +268,46 @@ public TaskStatus runTask(TaskToolbox toolbox) throws Exception | |
| ); | ||
| } | ||
|
|
||
| // Kill segments - order of steps 1, 2, 3, 4 must remain the same | ||
|
|
||
| // 1. Determine parent segment ids of killable unused segments | ||
| final Map<String, String> upgradedFromSegmentIds | ||
| = fetchParentIdsForSegments(toolbox, unusedIdToSegmentPlus); | ||
|
|
||
| // 2. Identify killable segments whose load specs are not shared with any other segment | ||
| final List<DataSegment> segmentsToKillFromDeepStore = getKillableSegments( | ||
| unusedIdToSegmentPlus, | ||
| upgradedFromSegmentIds, | ||
| usedSegmentLoadSpecs, | ||
| taskActionClient | ||
| ); | ||
|
|
||
| // 2a. Track segments that cannot be removed from deep store yet | ||
| final Set<DataSegment> segmentsNotKilled = new HashSet<>(unusedSegments); | ||
| segmentsToKillFromDeepStore.forEach(segmentsNotKilled::remove); | ||
| if (!segmentsNotKilled.isEmpty()) { | ||
| LOG.warn( | ||
| "Skipping kill of [%d] segments of datasource[%s] from deep storage" | ||
|
abhishekrb19 marked this conversation as resolved.
|
||
| + " as their load specs are shared by other segments.", | ||
|
abhishekrb19 marked this conversation as resolved.
|
||
| segmentsNotKilled.size(), getDataSource() | ||
| ); | ||
| } | ||
|
|
||
| // 3. Nuke all eligible unused segments | ||
|
Member
There was a problem hiding this comment. [P1] Close the sharing race before deleting deep storage The new relationship lookup is only a preflight before
Contributor
Author
There was a problem hiding this comment. It is very unlikely for this to happen. It had been called out here as well. Essentially, the kill buffer period protects us against this.
Contributor
Author
There was a problem hiding this comment. Adding a comment in javadoc. |
||
| taskActionClient.submit(new SegmentNukeAction(unusedSegments)); | ||
| emitMetric(toolbox.getEmitter(), TaskMetrics.SEGMENTS_DELETED_FROM_METADATA_STORE, unusedIdToSegmentPlus.size()); | ||
|
|
||
| // 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++; | ||
| numSegmentsKilled += segmentsToKillFromDeepStore.size(); | ||
|
|
||
| logInfo("Processed [%d] batches for kill task[%s].", numBatchesProcessed, getId()); | ||
|
|
||
| nextBatchSize = computeNextBatchSize(numSegmentsKilled); | ||
| } while (!unusedSegmentsPlus.isEmpty() && (null == numTotalBatches || numBatchesProcessed < numTotalBatches)); | ||
|
|
||
| final String taskId = getId(); | ||
| logInfo( | ||
|
|
@@ -342,7 +343,7 @@ int computeNextBatchSize(int numSegmentsKilled) | |
| /** | ||
| * Fetches the next batch of unused segments that are eligible for kill. | ||
| */ | ||
| protected List<DataSegmentPlus> fetchNextBatchOfUnusedSegments(TaskToolbox toolbox, int nextBatchSize) throws IOException | ||
| { | ||
| return toolbox.getTaskActionClient().submit( | ||
| new RetrieveUnusedSegmentsAction( | ||
|
|
@@ -352,7 +353,44 @@ protected List<DataSegment> fetchNextBatchOfUnusedSegments(TaskToolbox toolbox, | |
| nextBatchSize, | ||
| maxUsedStatusLastUpdatedTime | ||
| ) | ||
| ) | ||
| .stream() | ||
| .map(segment -> new DataSegmentPlus(segment, null, null, null, null, null, null, null)) | ||
| .collect(Collectors.toList()); | ||
| } | ||
|
|
||
| /** | ||
| * Fetches the parent IDs (if any) for the given unused segments. | ||
| * | ||
| * @param unusedIdToSegmentPlus Map containing unused segments whose parent IDs | ||
| * need to be fetched | ||
| * @return Map from segment ID to the segment ID from which | ||
| * it was upgraded. If an input segment was not upgraded from any other segment, | ||
| * it does not have an entry in the map. | ||
| */ | ||
| protected Map<String, String> fetchParentIdsForSegments( | ||
| TaskToolbox toolbox, | ||
| Map<String, DataSegmentPlus> unusedIdToSegmentPlus | ||
| ) | ||
| { | ||
| try { | ||
| return toolbox.getTaskActionClient().submit( | ||
| new RetrieveUpgradedFromSegmentIdsAction(getDataSource(), unusedIdToSegmentPlus.keySet()) | ||
| ).getUpgradedFromSegmentIds(); | ||
| } | ||
| catch (Exception e) { | ||
| // Do not proceed with killing these segments as we cannot be sure if their | ||
| // load spec is shared by any other segment or not. If load spec is shared, | ||
| // segment files cannot be deleted from deep store. If load spec is not | ||
| // shared, segments cannot be deleted from metadata store as that would | ||
| // leave deep store files orphaned, and they would never be cleaned up. | ||
| throw new ISE( | ||
| e, | ||
| "Could not retrieve parent segment ids using task action[retrieveUpgradedFromSegmentIds]." | ||
| + " Stopping kill task to avoid data loss in case the segment files" | ||
| + " are shared by other segments." | ||
|
Comment on lines
+388
to
+391
Contributor
There was a problem hiding this comment. What do you think about adding a unit test in
Contributor
Author
There was a problem hiding this comment. Thanks for the suggestion! Added some tests. |
||
| ); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
|
|
@@ -387,21 +425,20 @@ private NavigableMap<DateTime, List<TaskLock>> getNonRevokedTaskLockMap(TaskActi | |
| * @return list of segments to kill from deep storage | ||
| */ | ||
| private List<DataSegment> getKillableSegments( | ||
| Map<String, DataSegmentPlus> unusedSegments, | ||
| Map<String, String> upgradedFromSegmentIds, | ||
| Set<Map<String, Object>> usedSegmentLoadSpecs, | ||
| TaskActionClient taskActionClient | ||
| ) | ||
| { | ||
| // Determine parentId (or self, if no parent) for each unused segment | ||
| final Map<String, Set<DataSegment>> parentIdToUnusedSegments = new HashMap<>(); | ||
| for (Map.Entry<String, DataSegmentPlus> entry : unusedSegments.entrySet()) { | ||
| final String segmentId = entry.getKey(); | ||
| parentIdToUnusedSegments.computeIfAbsent( | ||
| upgradedFromSegmentIds.getOrDefault(segmentId, segmentId), | ||
| k -> new HashSet<>() | ||
| ).add(entry.getValue().getDataSegment()); | ||
| } | ||
|
|
||
| // Check if the parent or any of its children exist in metadata store | ||
|
|
@@ -411,10 +448,11 @@ private List<DataSegment> getKillableSegments( | |
| ); | ||
| if (response != null && response.getUpgradedToSegmentIds() != null) { | ||
| response.getUpgradedToSegmentIds().forEach((parent, children) -> { | ||
| if (!unusedSegments.keySet().containsAll(children)) { | ||
| // Do not kill segment if its load spec is shared by another segment | ||
| // which is not being killed. | ||
| LOG.info( | ||
| "Skipping kill of segments[%s] as its load spec is shared by segment IDs[%s].", | ||
| parentIdToUnusedSegments.get(parent), children | ||
| ); | ||
| parentIdToUnusedSegments.remove(parent); | ||
|
|
@@ -423,10 +461,14 @@ private List<DataSegment> getKillableSegments( | |
| } | ||
| } | ||
| catch (Exception e) { | ||
| // Do not proceed with the kill of any segment as we cannot be sure if their | ||
| // load specs are shared by any other segment | ||
| throw new ISE( | ||
| e, | ||
| "Could not perform task action[retrieveUpgradedToSegmentIds] to retrieve" | ||
| + " segment IDs which share load specs with segments being killed." | ||
| + " Stopping kill task to avoid data loss in case the segment files" | ||
| + " are shared by other segments." | ||
| ); | ||
| } | ||
|
|
||
|
|
@@ -449,7 +491,7 @@ private boolean isSegmentLoadSpecPresentIn( | |
| { | ||
| boolean isPresent = usedSegmentLoadSpecs.contains(segment.getLoadSpec()); | ||
| if (isPresent) { | ||
| LOG.info("Skipping kill of segment[%s] as its load spec is shared by other 'used' segments.", segment); | ||
| } | ||
| return isPresent; | ||
| } | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.