This is an automated email from the ASF dual-hosted git repository.
abhishekrb 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 96be82a3e67 Clean up duty for non-overlapping eternity tombstones
(#15281)
96be82a3e67 is described below
commit 96be82a3e67bb3fdb62d0640c0c8885f7f5813b0
Author: Abhishek Radhakrishnan <[email protected]>
AuthorDate: Mon Dec 11 08:57:15 2023 -0800
Clean up duty for non-overlapping eternity tombstones (#15281)
* Add initial draft of MarkDanglingTombstonesAsUnused duty.
* Use overshadowed segments instead of all used segments.
* Add unit test for MarkDanglingSegmentsAsUnused duty.
* Add mock call
* Simplify code.
* Docs
* shorter lines formatting
* metric doc
* More tests, refactor and fix up some logic.
* update javadocs; other review comments.
* Make numCorePartitions as 0 in the TombstoneShardSpec.
* fix up test
* Add tombstone core partition tests
* Update docs/design/coordinator.md
Co-authored-by: 317brian <[email protected]>
* review comment
* Minor cleanup
* Only consider tombstones with 0 core partitions
* Need to register the test shard type to make jackson happy
* test comments
* checkstyle
* fixup misc typos in comments
* Update logic to use overshadowed segments
* minor cleanup
* Rename duty to eternity tombstone instead of dangling. Add test for full
eternity tombstone.
* Address review feedback.
---------
Co-authored-by: 317brian <[email protected]>
---
docs/design/coordinator.md | 9 +-
docs/operations/metrics.md | 1 +
.../druid/testing/utils/MsqTestQueryHelper.java | 8 +-
.../timeline/partition/OverwriteShardSpec.java | 2 +-
.../org/apache/druid/discovery/BrokerClient.java | 2 +-
.../druid/server/coordinator/DruidCoordinator.java | 2 +
.../server/coordinator/duty/CompactSegments.java | 4 +-
.../duty/MarkEternityTombstonesAsUnused.java | 178 ++++++++
.../druid/server/coordinator/stats/Stats.java | 2 +
.../duty/MarkEternityTombstonesAsUnusedTest.java | 501 +++++++++++++++++++++
10 files changed, 701 insertions(+), 8 deletions(-)
diff --git a/docs/design/coordinator.md b/docs/design/coordinator.md
index e3652d2c344..9cb2279d497 100644
--- a/docs/design/coordinator.md
+++ b/docs/design/coordinator.md
@@ -61,7 +61,7 @@ org.apache.druid.cli.Main server coordinator
Segments can be automatically loaded and dropped from the cluster based on a
set of rules. For more information on rules, see [Rule
Configuration](../operations/rule-configuration.md).
-## Cleaning up segments
+### Clean up overshadowed segments
On each run, the Coordinator compares the set of used segments in the database
with the segments served by some
Historical nodes in the cluster. The Coordinator sends requests to Historical
nodes to unload unused segments or segments
@@ -70,6 +70,13 @@ that are removed from the database.
Segments that are overshadowed (their versions are too old and their data has
been replaced by newer segments) are
marked as unused. During the next Coordinator's run, they will be unloaded
from Historical nodes in the cluster.
+### Clean up non-overshadowed eternity tombstone segments
+
+On each run, the Coordinator determines and cleans up unneeded eternity
tombstone segments for each datasource. These segments must fit all the
following criteria:
+- It is a tombstone segment that starts at -INF or ends at INF (for example, a
tombstone with an interval of `-146136543-09-08T08:23:32.096Z/2000-01-01` or
`2020-01-01/146140482-04-24T15:36:27.903Z` or
`-146136543-09-08T08:23:32.096Z/146140482-04-24T15:36:27.903Z`)
+- It does not overlap with any overshadowed segment
+- It has 0 core partitions
+
## Segment availability
If a Historical service restarts or becomes unavailable for any reason, the
Coordinator will notice a service has gone missing and treat all segments
served by that service as being dropped. Given a sufficient period of time, the
segments may be reassigned to other Historical services in the cluster.
However, each segment that is dropped is not immediately forgotten. Instead,
there is a transitional data structure that stores all dropped segments with an
associated lifetime. The lifetime [...]
diff --git a/docs/operations/metrics.md b/docs/operations/metrics.md
index df8ce218bde..8fefc8b6e13 100644
--- a/docs/operations/metrics.md
+++ b/docs/operations/metrics.md
@@ -329,6 +329,7 @@ These metrics are for the Druid Coordinator and are reset
each time the Coordina
|`segment/size`|Total size of used segments in a data source. Emitted only for
data sources to which at least one used segment belongs.|`dataSource`|Varies|
|`segment/count`|Number of used segments belonging to a data source. Emitted
only for data sources to which at least one used segment
belongs.|`dataSource`|< max|
|`segment/overShadowed/count`|Number of segments marked as unused due to being
overshadowed.| |Varies|
+|`segment/unneededEternityTombstone/count`|Number of non-overshadowed eternity
tombstones marked as unused.| |Varies|
|`segment/unavailable/count`|Number of unique segments left to load until all
used segments are available for queries.|`dataSource`|0|
|`segment/underReplicated/count`|Number of segments, including replicas, left
to load until all used segments are available for queries.|`tier`,
`dataSource`|0|
|`tier/historical/count`|Number of available historical nodes in each
tier.|`tier`|Varies|
diff --git
a/integration-tests/src/main/java/org/apache/druid/testing/utils/MsqTestQueryHelper.java
b/integration-tests/src/main/java/org/apache/druid/testing/utils/MsqTestQueryHelper.java
index 7525cb9d874..a7510db860f 100644
---
a/integration-tests/src/main/java/org/apache/druid/testing/utils/MsqTestQueryHelper.java
+++
b/integration-tests/src/main/java/org/apache/druid/testing/utils/MsqTestQueryHelper.java
@@ -130,9 +130,11 @@ public class MsqTestQueryHelper extends
AbstractTestQueryHelper<MsqQueryWithResu
HttpResponseStatus httpResponseStatus = statusResponseHolder.getStatus();
if (!httpResponseStatus.equals(HttpResponseStatus.ACCEPTED)) {
throw new ISE(
- "Unable to submit the task successfully. Received response status
code [%d], and response content:\n[%s]",
- httpResponseStatus,
- statusResponseHolder.getContent()
+ StringUtils.format(
+ "Unable to submit the task successfully. Received response
status code [%d], and response content:\n[%s]",
+ httpResponseStatus.getCode(),
+ statusResponseHolder.getContent()
+ )
);
}
String content = statusResponseHolder.getContent();
diff --git
a/processing/src/main/java/org/apache/druid/timeline/partition/OverwriteShardSpec.java
b/processing/src/main/java/org/apache/druid/timeline/partition/OverwriteShardSpec.java
index f51e688c24c..ec784c44ffb 100644
---
a/processing/src/main/java/org/apache/druid/timeline/partition/OverwriteShardSpec.java
+++
b/processing/src/main/java/org/apache/druid/timeline/partition/OverwriteShardSpec.java
@@ -30,7 +30,7 @@ public interface OverwriteShardSpec extends ShardSpec
{
/**
* The core partition concept is not used with segment locking. Instead, the
{@link AtomicUpdateGroup} is used
- * to atomically overshadow segments. Here, we always returns 0 so that the
{@link PartitionHolder} skips checking
+ * to atomically overshadow segments. Here, we always return 0 so that the
{@link PartitionHolder} skips checking
* the completeness of the core partitions.
*/
@Override
diff --git a/server/src/main/java/org/apache/druid/discovery/BrokerClient.java
b/server/src/main/java/org/apache/druid/discovery/BrokerClient.java
index bc97c2490ef..a64c8d670e4 100644
--- a/server/src/main/java/org/apache/druid/discovery/BrokerClient.java
+++ b/server/src/main/java/org/apache/druid/discovery/BrokerClient.java
@@ -116,7 +116,7 @@ public class BrokerClient
catch (MalformedURLException e) {
// Not an IOException; this is our own fault.
throw DruidException.defensive(
- "Failed to build url with path[%] and query string [%s].",
+ "Failed to build url with path[%s] and query string [%s].",
oldRequest.getUrl().getPath(),
oldRequest.getUrl().getQuery()
);
diff --git
a/server/src/main/java/org/apache/druid/server/coordinator/DruidCoordinator.java
b/server/src/main/java/org/apache/druid/server/coordinator/DruidCoordinator.java
index 2415fcaaa24..5558d204e81 100644
---
a/server/src/main/java/org/apache/druid/server/coordinator/DruidCoordinator.java
+++
b/server/src/main/java/org/apache/druid/server/coordinator/DruidCoordinator.java
@@ -67,6 +67,7 @@ import org.apache.druid.server.coordinator.duty.KillRules;
import org.apache.druid.server.coordinator.duty.KillStalePendingSegments;
import org.apache.druid.server.coordinator.duty.KillSupervisors;
import org.apache.druid.server.coordinator.duty.KillUnusedSegments;
+import org.apache.druid.server.coordinator.duty.MarkEternityTombstonesAsUnused;
import
org.apache.druid.server.coordinator.duty.MarkOvershadowedSegmentsAsUnused;
import org.apache.druid.server.coordinator.duty.PrepareBalancerAndLoadQueues;
import org.apache.druid.server.coordinator.duty.RunRules;
@@ -515,6 +516,7 @@ public class DruidCoordinator
new UpdateReplicationStatus(),
new UnloadUnusedSegments(loadQueueManager),
new MarkOvershadowedSegmentsAsUnused(segments ->
metadataManager.segments().markSegmentsAsUnused(segments)),
+ new MarkEternityTombstonesAsUnused(segments ->
metadataManager.segments().markSegmentsAsUnused(segments)),
new BalanceSegments(config.getCoordinatorPeriod()),
new CollectSegmentAndServerStats(taskMaster)
);
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 ab92f180149..bb88b86dbf8 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
@@ -373,7 +373,7 @@ public class CompactSegments implements
CoordinatorCustomDuty
final String dataSourceName = segmentsToCompact.get(0).getDataSource();
- // As these segments will be compacted, we will aggregates the statistic
to the Compacted statistics
+ // As these segments will be compacted, we will aggregate the statistic
to the Compacted statistics
AutoCompactionSnapshot.Builder snapshotBuilder =
currentRunAutoCompactionSnapshotBuilders.computeIfAbsent(
dataSourceName,
k -> new AutoCompactionSnapshot.Builder(k,
AutoCompactionSnapshot.AutoCompactionScheduleStatus.RUNNING)
@@ -395,7 +395,7 @@ public class CompactSegments implements
CoordinatorCustomDuty
Granularity segmentGranularityToUse = null;
if (config.getGranularitySpec() == null ||
config.getGranularitySpec().getSegmentGranularity() == null) {
// Determines segmentGranularity from the segmentsToCompact
- // Each batch of segmentToCompact from CompactionSegmentIterator will
contains the same interval as
+ // 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()))) {
diff --git
a/server/src/main/java/org/apache/druid/server/coordinator/duty/MarkEternityTombstonesAsUnused.java
b/server/src/main/java/org/apache/druid/server/coordinator/duty/MarkEternityTombstonesAsUnused.java
new file mode 100644
index 00000000000..8a2e6c9dfd9
--- /dev/null
+++
b/server/src/main/java/org/apache/druid/server/coordinator/duty/MarkEternityTombstonesAsUnused.java
@@ -0,0 +1,178 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.server.coordinator.duty;
+
+import com.google.common.base.Optional;
+import org.apache.druid.client.DataSourcesSnapshot;
+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.common.logger.Logger;
+import org.apache.druid.server.coordinator.DruidCoordinatorRuntimeParams;
+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 org.apache.druid.timeline.DataSegment;
+import org.apache.druid.timeline.Partitions;
+import org.apache.druid.timeline.SegmentId;
+import org.apache.druid.timeline.SegmentTimeline;
+import org.apache.druid.timeline.partition.TombstoneShardSpec;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * Mark eternity tombstones not overshadowed by currently served segments as
unused. A candidate segment must fit all
+ * the criteria:
+ * <li> It is a tombstone that starts at {@link DateTimes#MIN} or ends at
{@link DateTimes#MAX} </li>
+ * <li> It does not overlap with any overshadowed segment in the datasource
</li>
+ * <li> It has has 0 core partitions i.e., {@link
TombstoneShardSpec#getNumCorePartitions()} == 0</li>
+ *
+ * <p>
+ * Only infinite-interval tombstones are considered as candidate segments in
this duty because they
+ * don't honor the preferred segment granularity specified at ingest time to
cover an underlying segment with
+ * {@link Granularities#ALL} as it can generate too many segments per time
chunk and cause an OOM. The infinite-interval
+ * tombstones make it hard to append data on the end of a data set that
started out with an {@link Granularities#ALL} eternity
+ * and then moved to actual time grains, so the compromise is that the
coordinator will remove these segments as long as it
+ * doesn't overlap any other segment.
+ * </p>
+ * <p>
+ * The overlapping condition is necessary as a candidate segment can overlap
with an overshadowed segment, and the latter
+ * needs to be marked as unused first by {@link
MarkOvershadowedSegmentsAsUnused} duty before the tombstone candidate
+ * can be marked as unused by {@link MarkEternityTombstonesAsUnused} duty.
+ *</p>
+ * <p>
+ * Only tombstones with 0 core partitions is considered as candidate segments.
Earlier generation tombstones with 1 core
+ * partition (i.e., {@link TombstoneShardSpec#getNumCorePartitions()} == 1)
are ignored by this duty because it can potentially
+ * cause data loss in a concurrent append and replace scenario and needs to be
manually cleaned up. See this
+ * <a href="https://github.com/apache/druid/pull/15379";>for details</a>.
+ * </p>
+ */
+public class MarkEternityTombstonesAsUnused implements CoordinatorDuty
+{
+ private static final Logger log = new
Logger(MarkEternityTombstonesAsUnused.class);
+
+ private final SegmentDeleteHandler deleteHandler;
+
+ public MarkEternityTombstonesAsUnused(final SegmentDeleteHandler
deleteHandler)
+ {
+ this.deleteHandler = deleteHandler;
+ }
+
+ @Override
+ public DruidCoordinatorRuntimeParams run(final DruidCoordinatorRuntimeParams
params)
+ {
+ DataSourcesSnapshot dataSourcesSnapshot = params.getDataSourcesSnapshot();
+
+ final Map<String, Set<SegmentId>>
datasourceToNonOvershadowedEternityTombstones =
+ determineNonOvershadowedEternityTombstones(
+ dataSourcesSnapshot
+ );
+
+ if (datasourceToNonOvershadowedEternityTombstones.size() == 0) {
+ log.debug("No non-overshadowed eternity tombstones found.");
+ return params;
+ }
+
+ log.debug("Found [%d] datasource containing non-overshadowed eternity
tombstones[%s]",
+ datasourceToNonOvershadowedEternityTombstones.size(),
datasourceToNonOvershadowedEternityTombstones
+ );
+
+ final CoordinatorRunStats stats = params.getCoordinatorStats();
+ datasourceToNonOvershadowedEternityTombstones.forEach((datasource,
nonOvershadowedEternityTombstones) -> {
+ final RowKey datasourceKey = RowKey.of(Dimension.DATASOURCE, datasource);
+ stats.add(Stats.Segments.UNNEEDED_ETERNITY_TOMBSTONE, datasourceKey,
nonOvershadowedEternityTombstones.size());
+ final int unusedCount =
deleteHandler.markSegmentsAsUnused(nonOvershadowedEternityTombstones);
+ log.info(
+ "Successfully marked [%d] non-overshadowed eternity tombstones[%s]
of datasource[%s] as unused.",
+ unusedCount,
+ nonOvershadowedEternityTombstones,
+ datasource
+ );
+ });
+
+ return params;
+ }
+
+ /**
+ * Computes the set of unneeded eternity tombstones per datasource using the
datasources snapshot. The computation is
+ * as follows:
+ *
+ * <li> Determine the set of used and non-overshadowed segments from the
used segments' timeline. </li>
+ * <li> For each such candidate segment that is a tombstone with an infinite
start or end, look at the set of overshadowed
+ * segments to see if any of the intervals overlaps with the candidate
segment.
+ * <li> If there is no overlap, add the candidate segment to the eternity
segments result set. </li>
+ * There can at most be two such candidate tombstones per datasource -- one
that starts at {@link DateTimes#MIN}
+ * and another that ends at {@link DateTimes#MAX}. </li>
+ * </p>
+ *
+ * @param dataSourcesSnapshot the datasources snapshot for segments timeline
+ * @return the set of non-overshadowed eternity tombstones grouped by
datasource
+ */
+ private Map<String, Set<SegmentId>>
determineNonOvershadowedEternityTombstones(final DataSourcesSnapshot
dataSourcesSnapshot)
+ {
+ final Map<String, Set<SegmentId>>
datasourceToNonOvershadowedEternityTombstones = new HashMap<>();
+
+ dataSourcesSnapshot.getDataSourcesMap().keySet().forEach((datasource) -> {
+ final SegmentTimeline usedSegmentsTimeline
+ =
dataSourcesSnapshot.getUsedSegmentsTimelinesPerDataSource().get(datasource);
+
+ final Optional<Set<DataSegment>> usedNonOvershadowedSegments =
+ Optional.fromNullable(usedSegmentsTimeline)
+ .transform(timeline ->
timeline.findNonOvershadowedObjectsInInterval(
+ Intervals.ETERNITY,
+ Partitions.ONLY_COMPLETE
+ ));
+
+ if (usedNonOvershadowedSegments.isPresent()) {
+ usedNonOvershadowedSegments.get().forEach(candidateSegment -> {
+ if (isNewGenerationEternityTombstone(candidateSegment)) {
+ boolean overlaps =
dataSourcesSnapshot.getOvershadowedSegments().stream()
+ .filter(overshadowedSegment
->
+
candidateSegment.getDataSource()
+
.equals(overshadowedSegment.getDataSource()))
+ .anyMatch(
+ overshadowedSegment ->
+
candidateSegment.getInterval()
+
.overlaps(overshadowedSegment.getInterval())
+ );
+ if (!overlaps) {
+ datasourceToNonOvershadowedEternityTombstones
+ .computeIfAbsent(datasource, ds -> new HashSet<>())
+ .add(candidateSegment.getId());
+ }
+ }
+ });
+ }
+ });
+
+ return datasourceToNonOvershadowedEternityTombstones;
+ }
+
+ private boolean isNewGenerationEternityTombstone(final DataSegment segment)
+ {
+ return segment.isTombstone() &&
segment.getShardSpec().getNumCorePartitions() == 0 && (
+ DateTimes.MIN.equals(segment.getInterval().getStart()) ||
+ DateTimes.MAX.equals(segment.getInterval().getEnd()));
+ }
+}
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 539d58d5594..a964a28aab7 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
@@ -57,6 +57,8 @@ public class Stats
= CoordinatorStat.toDebugAndEmit("unneeded", "segment/unneeded/count");
public static final CoordinatorStat OVERSHADOWED
= CoordinatorStat.toDebugAndEmit("overshadowed",
"segment/overshadowed/count");
+ public static final CoordinatorStat UNNEEDED_ETERNITY_TOMBSTONE
+ = CoordinatorStat.toDebugAndEmit("unneededEternityTombstone",
"segment/unneededEternityTombstone/count");
}
public static class SegmentQueue
diff --git
a/server/src/test/java/org/apache/druid/server/coordinator/duty/MarkEternityTombstonesAsUnusedTest.java
b/server/src/test/java/org/apache/druid/server/coordinator/duty/MarkEternityTombstonesAsUnusedTest.java
new file mode 100644
index 00000000000..d6dd4f98206
--- /dev/null
+++
b/server/src/test/java/org/apache/druid/server/coordinator/duty/MarkEternityTombstonesAsUnusedTest.java
@@ -0,0 +1,501 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.server.coordinator.duty;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Sets;
+import org.apache.druid.client.DruidServer;
+import org.apache.druid.java.util.common.DateTimes;
+import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.server.coordination.ServerType;
+import org.apache.druid.server.coordinator.CoordinatorDynamicConfig;
+import org.apache.druid.server.coordinator.DruidCluster;
+import org.apache.druid.server.coordinator.DruidCoordinatorRuntimeParams;
+import org.apache.druid.server.coordinator.ServerHolder;
+import org.apache.druid.server.coordinator.balancer.RandomBalancerStrategy;
+import org.apache.druid.server.coordinator.loading.SegmentLoadQueueManager;
+import org.apache.druid.server.coordinator.loading.TestLoadQueuePeon;
+import
org.apache.druid.server.coordinator.simulate.TestSegmentsMetadataManager;
+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 org.apache.druid.timeline.DataSegment;
+import org.apache.druid.timeline.SegmentTimeline;
+import org.apache.druid.timeline.partition.TombstoneShardSpec;
+import org.joda.time.Interval;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.Set;
+
+public class MarkEternityTombstonesAsUnusedTest
+{
+ private final String ds1 = "foo";
+ private final String ds2 = "bar";
+
+ // The verbose variable names follow the convention for readability in the
tests:
+ // datasource name - shard spec type - interval - version
+ private final DataSegment ds1NumberedSegmentMinToMaxV0 =
DataSegment.builder().dataSource(ds1)
+
.interval(Intervals.ETERNITY)
+
.version("0")
+ .size(0)
+ .build();
+
+ private final DataSegment ds1TombstoneSegmentMinToMaxV1 =
DataSegment.builder().dataSource(ds1)
+
.shardSpec(new TombstoneShardSpec())
+
.interval(Intervals.ETERNITY)
+
.version("1")
+ .size(0)
+
.build();
+
+ private final DataSegment ds1TombstoneSegmentMinTo2000V1 =
DataSegment.builder().dataSource(ds1)
+
.shardSpec(new TombstoneShardSpec())
+
.interval(new Interval(DateTimes.MIN, DateTimes.of("2000")))
+
.version("1")
+
.size(0)
+
.build();
+
+ private final DataSegment ds1NumberedSegment2000To2001V1 =
DataSegment.builder().dataSource(ds1)
+
.interval(Intervals.of("2000/2001"))
+
.version("1")
+
.size(0)
+
.build();
+
+ private final DataSegment ds1TombstoneSegment2001ToMaxV1 =
DataSegment.builder().dataSource(ds1)
+
.shardSpec(new TombstoneShardSpec())
+
.interval(new Interval(DateTimes.of("2001"), DateTimes.MAX))
+
.version("1")
+
.size(0)
+
.build();
+
+ final DataSegment ds1TombstoneSegmentMinTo2000V2 =
ds1TombstoneSegmentMinTo2000V1.withVersion("2");
+ final DataSegment ds1TombstoneSegment2001ToMaxV2 =
ds1TombstoneSegment2001ToMaxV1.withVersion("2");
+
+ private final DataSegment ds2TombstoneSegment1995To2005V0 =
DataSegment.builder().dataSource(ds2)
+
.shardSpec(new TombstoneShardSpec())
+
.interval(Intervals.of("1995/2005"))
+
.version("0")
+
.size(0)
+
.build();
+
+ private final DataSegment ds2TombstoneSegmentMinTo2000V1 =
DataSegment.builder().dataSource(ds2)
+
.shardSpec(new TombstoneShardSpec())
+
.interval(new Interval(DateTimes.MIN, DateTimes.of("2000")))
+
.version("1")
+
.size(0)
+
.build();
+
+ private final DataSegment ds2NumberedSegment3000To4000V1 =
DataSegment.builder().dataSource(ds2)
+
.interval(Intervals.of("3000/4000"))
+
.version("1")
+
.size(0)
+
.build();
+
+ private final DataSegment ds2TombstoneSegment4000ToMaxV1 =
DataSegment.builder().dataSource(ds2)
+
.shardSpec(new TombstoneShardSpec())
+
.interval(new Interval(DateTimes.of("4000"), DateTimes.MAX))
+
.version("1")
+
.size(0)
+
.build();
+
+ private final DataSegment ds2TombstoneSegment4000To4001V1 =
DataSegment.builder().dataSource(ds2)
+
.shardSpec(new TombstoneShardSpec())
+
.interval(new Interval(DateTimes.of("4000"), DateTimes.of("4001")))
+
.version("1")
+
.size(0)
+
.build();
+
+ private final DataSegment ds2NumberedSegment1999To2500V2 =
DataSegment.builder().dataSource(ds2)
+
.interval(Intervals.of("1999/2500"))
+
.version("2")
+
.size(0)
+
.build();
+
+
+ /**
+ * An old generation tombstone with 1 core partition instead of the default
0.
+ */
+ private final DataSegment ds2TombstoneSegment4000ToMaxV1With1CorePartition =
ds2TombstoneSegment4000ToMaxV1.withShardSpec(
+ new TombstoneShardSpec() {
+ @Override
+ @JsonProperty("partitions")
+ public int getNumCorePartitions()
+ {
+ return 1;
+ }
+ });
+
+
+ private TestSegmentsMetadataManager segmentsMetadataManager;
+
+ @Before
+ public void setup()
+ {
+ segmentsMetadataManager = new TestSegmentsMetadataManager();
+ }
+
+ /**
+ * Half-inifinity tombstones overlapping with overshadowed segments
shouldn't be marked as unused.
+ */
+ @Test
+ public void testCandidateTombstonesWithUsedOvershadowedSegments()
+ {
+ final ImmutableList<DataSegment> allUsedSegments = ImmutableList.of(
+ ds1NumberedSegmentMinToMaxV0,
+ ds1TombstoneSegmentMinTo2000V1,
+ ds1NumberedSegment2000To2001V1,
+ ds1TombstoneSegment2001ToMaxV1
+ );
+ final ImmutableList<DataSegment> expectedUsedSegments =
ImmutableList.copyOf(allUsedSegments);
+ final DruidCoordinatorRuntimeParams params =
initializeServerAndGetParams(allUsedSegments);
+
+ final SegmentTimeline timeline =
segmentsMetadataManager.getSnapshotOfDataSourcesWithAllUsedSegments()
+
.getUsedSegmentsTimelinesPerDataSource()
+ .get(ds1);
+
+ // Verify that the half-infinity tombstone is overshadowed and everything
else is not
+ Assert.assertTrue(timeline.isOvershadowed(ds1NumberedSegmentMinToMaxV0));
+
Assert.assertFalse(timeline.isOvershadowed(ds1TombstoneSegmentMinTo2000V1));
+
Assert.assertFalse(timeline.isOvershadowed(ds1NumberedSegment2000To2001V1));
+
Assert.assertFalse(timeline.isOvershadowed(ds1TombstoneSegment2001ToMaxV1));
+
+ runEternityTombstonesDutyAndVerify(params, allUsedSegments,
expectedUsedSegments);
+ }
+
+ /**
+ * Inifinity tombstones overlapping with overshadowed segments shouldn't be
marked as unused.
+ */
+ @Test
+ public void testCandidateTombstonesWithUsedOvershadowedSegments2()
+ {
+ final ImmutableList<DataSegment> allUsedSegments = ImmutableList.of(
+ ds1NumberedSegmentMinToMaxV0,
+ ds1TombstoneSegmentMinToMaxV1
+ );
+ final ImmutableList<DataSegment> expectedUsedSegments =
ImmutableList.copyOf(allUsedSegments);
+ final DruidCoordinatorRuntimeParams params =
initializeServerAndGetParams(allUsedSegments);
+
+ final SegmentTimeline ds1Timeline =
segmentsMetadataManager.getSnapshotOfDataSourcesWithAllUsedSegments()
+
.getUsedSegmentsTimelinesPerDataSource()
+ .get(ds1);
+
+
Assert.assertTrue(ds1Timeline.isOvershadowed(ds1NumberedSegmentMinToMaxV0));
+
Assert.assertFalse(ds1Timeline.isOvershadowed(ds1TombstoneSegmentMinToMaxV1));
+
+ runEternityTombstonesDutyAndVerify(params, allUsedSegments,
expectedUsedSegments);
+ }
+
+ /**
+ * Half-inifinity tombstones that don't overlap with an overshadowed segment
should be marked as unused.
+ */
+ @Test
+ public void testCandidateTombstonesWithNoUsedOvershadowedSegments()
+ {
+ final ImmutableList<DataSegment> allUsedSegments = ImmutableList.of(
+ ds1TombstoneSegmentMinTo2000V1,
+ ds1NumberedSegment2000To2001V1,
+ ds1TombstoneSegment2001ToMaxV1
+ );
+ final ImmutableList<DataSegment> expectedUsedSegments = ImmutableList.of(
+ ds1NumberedSegment2000To2001V1
+ );
+ final DruidCoordinatorRuntimeParams params =
initializeServerAndGetParams(allUsedSegments);
+
+ SegmentTimeline timeline =
segmentsMetadataManager.getSnapshotOfDataSourcesWithAllUsedSegments()
+
.getUsedSegmentsTimelinesPerDataSource()
+ .get(ds1);
+
+
Assert.assertFalse(timeline.isOvershadowed(ds1TombstoneSegmentMinTo2000V1));
+
Assert.assertFalse(timeline.isOvershadowed(ds1NumberedSegment2000To2001V1));
+
Assert.assertFalse(timeline.isOvershadowed(ds1TombstoneSegmentMinTo2000V2));
+
+ runEternityTombstonesDutyAndVerify(params, allUsedSegments,
expectedUsedSegments);
+ }
+
+ /**
+ * Full-infinity tombstones that has no overlap should be marked as unused.
+ */
+ @Test
+ public void testCandidateTombstonesWithNoUsedOvershadowedSegments2()
+ {
+ final ImmutableList<DataSegment> allUsedSegments = ImmutableList.of(
+ ds1TombstoneSegmentMinToMaxV1
+ );
+ final ImmutableList<DataSegment> expectedUsedSegments = ImmutableList.of();
+ final DruidCoordinatorRuntimeParams params =
initializeServerAndGetParams(allUsedSegments);
+
+ final SegmentTimeline ds1Timeline =
segmentsMetadataManager.getSnapshotOfDataSourcesWithAllUsedSegments()
+
.getUsedSegmentsTimelinesPerDataSource()
+ .get(ds1);
+
+
Assert.assertFalse(ds1Timeline.isOvershadowed(ds1TombstoneSegmentMinToMaxV1));
+
+ runEternityTombstonesDutyAndVerify(params, allUsedSegments,
expectedUsedSegments);
+ }
+
+ /**
+ * Half-inifinity tombstones that overlap with overshadowed used segments
shouldn't be marked as unused.
+ */
+ @Test
+ public void testCandiateTombstonesWithManyOvershadowedSegments()
+ {
+ final ImmutableList<DataSegment> allUsedSegments = ImmutableList.of(
+ ds1TombstoneSegmentMinTo2000V1,
+ ds1NumberedSegment2000To2001V1,
+ ds1TombstoneSegment2001ToMaxV1,
+ ds1TombstoneSegmentMinTo2000V2,
+ ds1TombstoneSegment2001ToMaxV2
+ );
+ final ImmutableList<DataSegment> expectedUsedSegments =
ImmutableList.copyOf(allUsedSegments);
+
+ final DruidCoordinatorRuntimeParams params =
initializeServerAndGetParams(allUsedSegments);
+
+ SegmentTimeline timeline =
segmentsMetadataManager.getSnapshotOfDataSourcesWithAllUsedSegments()
+
.getUsedSegmentsTimelinesPerDataSource()
+ .get(ds1);
+
+ Assert.assertTrue(timeline.isOvershadowed(ds1TombstoneSegmentMinTo2000V1));
+ Assert.assertTrue(timeline.isOvershadowed(ds1TombstoneSegment2001ToMaxV1));
+
Assert.assertFalse(timeline.isOvershadowed(ds1NumberedSegment2000To2001V1));
+
Assert.assertFalse(timeline.isOvershadowed(ds1TombstoneSegmentMinTo2000V2));
+
Assert.assertFalse(timeline.isOvershadowed(ds1TombstoneSegment2001ToMaxV2));
+
+ runEternityTombstonesDutyAndVerify(params, allUsedSegments,
expectedUsedSegments);
+ }
+
+ /**
+ * <p>
+ * Datasource 1 has the following half-infinity tombstones:
+ * <li> {@link #ds1TombstoneSegmentMinTo2000V1} is overshadowed by {@link
#ds1TombstoneSegmentMinTo2000V2}, so cannot be marked as unused. </li>
+ * <li> {@link #ds1TombstoneSegmentMinTo2000V2} overlaps with {@link
#ds1TombstoneSegmentMinTo2000V1}, so cannot be marked as unused. </li>
+ * <li> {@link #ds1TombstoneSegment2001ToMaxV1} doesn't overlap with any
other segment and can be marked as unused. </li>
+ *
+ * Note that {@link #ds1TombstoneSegmentMinTo2000V1} will be marked as
unused by {@link MarkOvershadowedSegmentsAsUnused} duty
+ * and then subsequently {@link #ds1TombstoneSegmentMinTo2000V2} will be
marked as unused by the {@link MarkEternityTombstonesAsUnused}
+ * duty eventually.
+ * </p>
+ *
+ * <p>
+ * Datasource 2 has half eternity tombstones that don't overlap with any
other segment, so both can be removed.
+ * </p>
+ */
+ @Test
+ public void testCandidateTombstonesInMultipleDatasources()
+ {
+ final ImmutableList<DataSegment> allUsedSegments = ImmutableList.of(
+ ds1TombstoneSegmentMinTo2000V1,
+ ds1NumberedSegment2000To2001V1,
+ ds1TombstoneSegment2001ToMaxV1,
+ ds1TombstoneSegmentMinTo2000V2,
+ ds2TombstoneSegmentMinTo2000V1,
+ ds2NumberedSegment3000To4000V1,
+ ds2TombstoneSegment4000ToMaxV1
+ );
+
+ final ImmutableList<DataSegment> expectedUsedSegments = ImmutableList.of(
+ ds1TombstoneSegmentMinTo2000V1,
+ ds1NumberedSegment2000To2001V1,
+ ds1TombstoneSegmentMinTo2000V2,
+ ds2NumberedSegment3000To4000V1
+ );
+ final DruidCoordinatorRuntimeParams params =
initializeServerAndGetParams(allUsedSegments);
+
+ SegmentTimeline ds1Timeline =
segmentsMetadataManager.getSnapshotOfDataSourcesWithAllUsedSegments()
+
.getUsedSegmentsTimelinesPerDataSource()
+ .get(ds1);
+
Assert.assertTrue(ds1Timeline.isOvershadowed(ds1TombstoneSegmentMinTo2000V1));
+
Assert.assertFalse(ds1Timeline.isOvershadowed(ds1NumberedSegment2000To2001V1));
+
Assert.assertFalse(ds1Timeline.isOvershadowed(ds1TombstoneSegment2001ToMaxV1));
+
Assert.assertFalse(ds1Timeline.isOvershadowed(ds1TombstoneSegmentMinTo2000V2));
+
+ SegmentTimeline ds2Timeline =
segmentsMetadataManager.getSnapshotOfDataSourcesWithAllUsedSegments()
+
.getUsedSegmentsTimelinesPerDataSource()
+ .get(ds2);
+
Assert.assertFalse(ds2Timeline.isOvershadowed(ds2TombstoneSegmentMinTo2000V1));
+
Assert.assertFalse(ds2Timeline.isOvershadowed(ds2NumberedSegment3000To4000V1));
+
Assert.assertFalse(ds2Timeline.isOvershadowed(ds2TombstoneSegment4000ToMaxV1));
+
+ runEternityTombstonesDutyAndVerify(params, allUsedSegments,
expectedUsedSegments);
+ }
+
+ /**
+ * Half-inifinity tombstones that partially overlaps with other segments
(not overshadowed) can still
+ * be marked as used.
+ */
+ @Test
+ public void testCandidateTombstonesWithPartiallyOverlappingSegment()
+ {
+ final ImmutableList<DataSegment> allUsedSegments = ImmutableList.of(
+ ds2TombstoneSegment1995To2005V0,
+ ds2TombstoneSegmentMinTo2000V1,
+ ds2NumberedSegment3000To4000V1,
+ ds2TombstoneSegment4000ToMaxV1
+ );
+
+ final ImmutableList<DataSegment> expectedUsedSegments = ImmutableList.of(
+ ds2TombstoneSegment1995To2005V0,
+ ds2NumberedSegment3000To4000V1
+ );
+ final DruidCoordinatorRuntimeParams params =
initializeServerAndGetParams(allUsedSegments);
+
+ final SegmentTimeline ds2Timeline =
segmentsMetadataManager.getSnapshotOfDataSourcesWithAllUsedSegments()
+
.getUsedSegmentsTimelinesPerDataSource()
+ .get(ds2);
+
+
Assert.assertFalse(ds2Timeline.isOvershadowed(ds2TombstoneSegment1995To2005V0));
+
Assert.assertFalse(ds2Timeline.isOvershadowed(ds2TombstoneSegmentMinTo2000V1));
+
Assert.assertFalse(ds2Timeline.isOvershadowed(ds2NumberedSegment3000To4000V1));
+
Assert.assertFalse(ds2Timeline.isOvershadowed(ds2TombstoneSegment4000ToMaxV1));
+
+ runEternityTombstonesDutyAndVerify(params, allUsedSegments,
expectedUsedSegments);
+ }
+
+ /**
+ * Half-inifinity tombstones that partially overlaps with other used
segments would still be considered as used and
+ * non-overshadowed and can be marked as unused.
+ */
+ @Test
+ public void
testCandidateTombstonesWithPartiallyOverlappingHigherVersionUsedSegment()
+ {
+ final ImmutableList<DataSegment> allUsedSegments = ImmutableList.of(
+ ds2TombstoneSegmentMinTo2000V1,
+ ds2NumberedSegment3000To4000V1,
+ ds2TombstoneSegment4000ToMaxV1,
+ ds2NumberedSegment1999To2500V2
+ );
+
+ final ImmutableList<DataSegment> expectedUsedSegments = ImmutableList.of(
+ ds2NumberedSegment3000To4000V1,
+ ds2NumberedSegment1999To2500V2
+ );
+
+ final DruidCoordinatorRuntimeParams params =
initializeServerAndGetParams(allUsedSegments);
+
+ final SegmentTimeline ds2Timeline =
segmentsMetadataManager.getSnapshotOfDataSourcesWithAllUsedSegments()
+
.getUsedSegmentsTimelinesPerDataSource()
+ .get(ds2);
+
Assert.assertFalse(ds2Timeline.isOvershadowed(ds2TombstoneSegmentMinTo2000V1));
+
Assert.assertFalse(ds2Timeline.isOvershadowed(ds2NumberedSegment3000To4000V1));
+
Assert.assertFalse(ds2Timeline.isOvershadowed(ds2TombstoneSegment4000ToMaxV1));
+
Assert.assertFalse(ds2Timeline.isOvershadowed(ds2NumberedSegment1999To2500V2));
+
+ runEternityTombstonesDutyAndVerify(params, allUsedSegments,
expectedUsedSegments);
+ }
+
+ /**
+ * Finite-interval tombstones shouldn't be marked as unused.
+ */
+ @Test
+ public void testFiniteIntervalTombstone()
+ {
+ final ImmutableList<DataSegment> allUsedSegments = ImmutableList.of(
+ ds2TombstoneSegment4000To4001V1
+ );
+ final ImmutableList<DataSegment> expectedUsedSegments = ImmutableList.of(
+ ds2TombstoneSegment4000To4001V1
+ );
+ final DruidCoordinatorRuntimeParams params =
initializeServerAndGetParams(allUsedSegments);
+
+ final SegmentTimeline ds2Timeline =
segmentsMetadataManager.getSnapshotOfDataSourcesWithAllUsedSegments()
+
.getUsedSegmentsTimelinesPerDataSource()
+ .get(ds2);
+
+
Assert.assertFalse(ds2Timeline.isOvershadowed(ds2TombstoneSegment4000To4001V1));
+
+ runEternityTombstonesDutyAndVerify(params, allUsedSegments,
expectedUsedSegments);
+ }
+
+ /**
+ * Tombstones with 1 core partition i.e., {@link
TombstoneShardSpec#getNumCorePartitions()} == 1 shouldn't be
+ * marked as unused.
+ */
+ @Test
+ public void testTombstoneWith1CorePartition()
+ {
+ final ImmutableList<DataSegment> allUsedSegments = ImmutableList.of(
+ ds2TombstoneSegment4000ToMaxV1With1CorePartition
+ );
+ final ImmutableList<DataSegment> expectedUsedSegments = ImmutableList.of(
+ ds2TombstoneSegment4000ToMaxV1With1CorePartition
+ );
+ final DruidCoordinatorRuntimeParams params =
initializeServerAndGetParams(allUsedSegments);
+
+ final SegmentTimeline ds2Timeline =
segmentsMetadataManager.getSnapshotOfDataSourcesWithAllUsedSegments()
+
.getUsedSegmentsTimelinesPerDataSource()
+ .get(ds2);
+
+
Assert.assertFalse(ds2Timeline.isOvershadowed(ds2TombstoneSegment4000ToMaxV1With1CorePartition));
+
+ runEternityTombstonesDutyAndVerify(params, allUsedSegments,
expectedUsedSegments);
+ }
+
+ private DruidCoordinatorRuntimeParams initializeServerAndGetParams(final
ImmutableList<DataSegment> segments)
+ {
+ final DruidServer druidServer = new DruidServer("", "", "", 0L,
ServerType.fromString("broker"), "", 0);
+ for (final DataSegment segment : segments) {
+ segmentsMetadataManager.addSegment(segment);
+ druidServer.addDataSegment(segment);
+ }
+
+ final DruidCluster druidCluster = DruidCluster
+ .builder()
+ .add(new ServerHolder(druidServer.toImmutableDruidServer(), new
TestLoadQueuePeon()))
+ .build();
+
+ final DruidCoordinatorRuntimeParams params = DruidCoordinatorRuntimeParams
+ .newBuilder(DateTimes.nowUtc())
+ .withDataSourcesSnapshot(
+
segmentsMetadataManager.getSnapshotOfDataSourcesWithAllUsedSegments()
+ )
+ .withDruidCluster(druidCluster)
+ .withDynamicConfigs(
+
CoordinatorDynamicConfig.builder().withMarkSegmentAsUnusedDelayMillis(0).build()
+ )
+ .withBalancerStrategy(new RandomBalancerStrategy())
+ .withSegmentAssignerUsing(new SegmentLoadQueueManager(null, null))
+ .build();
+
+ return params;
+ }
+
+ private void runEternityTombstonesDutyAndVerify(
+ DruidCoordinatorRuntimeParams params,
+ final ImmutableList<DataSegment> allUsedSegments,
+ final ImmutableList<DataSegment> expectedUsedSegments
+ )
+ {
+ params = new
MarkEternityTombstonesAsUnused(segmentsMetadataManager::markSegmentsAsUnused).run(params);
+
+ final Set<DataSegment> actualUsedSegments =
Sets.newHashSet(segmentsMetadataManager.iterateAllUsedSegments());
+
+ Assert.assertEquals(expectedUsedSegments.size(),
actualUsedSegments.size());
+ Assert.assertTrue(actualUsedSegments.containsAll(expectedUsedSegments));
+
+ final CoordinatorRunStats runStats = params.getCoordinatorStats();
+ Assert.assertEquals(
+ allUsedSegments.size() - expectedUsedSegments.size(),
+ runStats.get(Stats.Segments.UNNEEDED_ETERNITY_TOMBSTONE,
RowKey.of(Dimension.DATASOURCE, ds1)) +
+ runStats.get(Stats.Segments.UNNEEDED_ETERNITY_TOMBSTONE,
RowKey.of(Dimension.DATASOURCE, ds2))
+ );
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
