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 d0f3fdab370 Allow using different lock types for kill task, remove
markAsUnused parameter (#16362)
d0f3fdab370 is described below
commit d0f3fdab370aff104acc4e69a6e90a9a76ffb08b
Author: Igor Berman <[email protected]>
AuthorDate: Fri May 10 04:07:36 2024 +0300
Allow using different lock types for kill task, remove markAsUnused
parameter (#16362)
Changes:
- Remove deprecated `markAsUnused` parameter from `KillUnusedSegmentsTask`
- Allow `kill` task to use `REPLACE` lock when `useConcurrentLocks` is true
- Use `EXCLUSIVE` lock by default
---
.../indexing/client/ControllerChatHandlerTest.java | 2 +-
.../common/task/KillUnusedSegmentsTask.java | 83 +++---
...ClientKillUnusedSegmentsTaskQuerySerdeTest.java | 8 -
.../common/task/KillUnusedSegmentsTaskTest.java | 285 +++++++++++++--------
.../batch/parallel/PushedSegmentsReportTest.java | 2 +-
.../druid/indexing/overlord/TaskLifecycleTest.java | 2 -
.../overlord/http/OverlordResourceTest.java | 2 +-
.../druid/indexer/report/KillTaskReport.java | 19 +-
.../druid/indexer/report/TaskReportSerdeTest.java | 3 +-
.../ClientKillUnusedSegmentsTaskQuery.java | 20 +-
.../apache/druid/rpc/indexing/OverlordClient.java | 1 -
.../ClientKillUnusedSegmentsTaskQueryTest.java | 8 -
.../druid/rpc/indexing/OverlordClientImplTest.java | 1 -
13 files changed, 223 insertions(+), 213 deletions(-)
diff --git
a/extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/indexing/client/ControllerChatHandlerTest.java
b/extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/indexing/client/ControllerChatHandlerTest.java
index 93436c84ead..dd04814d879 100644
---
a/extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/indexing/client/ControllerChatHandlerTest.java
+++
b/extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/indexing/client/ControllerChatHandlerTest.java
@@ -41,7 +41,7 @@ public class ControllerChatHandlerTest
final Controller controller = Mockito.mock(Controller.class);
TaskReport.ReportMap reportMap = new TaskReport.ReportMap();
- reportMap.put("killUnusedSegments", new KillTaskReport("kill_1", new
KillTaskReport.Stats(1, 2, 3)));
+ reportMap.put("killUnusedSegments", new KillTaskReport("kill_1", new
KillTaskReport.Stats(1, 2)));
Mockito.when(controller.liveReports())
.thenReturn(reportMap);
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 78a7abae7a6..6a1f9e95a06 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
@@ -32,13 +32,14 @@ import org.apache.druid.indexer.TaskStatus;
import org.apache.druid.indexer.report.KillTaskReport;
import org.apache.druid.indexer.report.TaskReport;
import org.apache.druid.indexing.common.TaskLock;
+import org.apache.druid.indexing.common.TaskLockType;
import org.apache.druid.indexing.common.TaskToolbox;
-import org.apache.druid.indexing.common.actions.MarkSegmentsAsUnusedAction;
import org.apache.druid.indexing.common.actions.RetrieveUnusedSegmentsAction;
import org.apache.druid.indexing.common.actions.RetrieveUsedSegmentsAction;
import org.apache.druid.indexing.common.actions.SegmentNukeAction;
import org.apache.druid.indexing.common.actions.TaskActionClient;
import org.apache.druid.indexing.common.actions.TaskLocks;
+import org.apache.druid.indexing.common.actions.TimeChunkLockTryAcquireAction;
import org.apache.druid.indexing.overlord.Segments;
import org.apache.druid.java.util.common.ISE;
import org.apache.druid.java.util.common.StringUtils;
@@ -46,7 +47,6 @@ import org.apache.druid.java.util.common.logger.Logger;
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;
@@ -66,9 +66,6 @@ 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.
- * <p>
- * The field {@link #isMarkAsUnused()} is now deprecated.
- * </p>
*/
public class KillUnusedSegmentsTask extends AbstractFixedIntervalTask
{
@@ -91,8 +88,6 @@ public class KillUnusedSegmentsTask extends
AbstractFixedIntervalTask
@Nullable
private final List<String> versions;
- @Deprecated
- private final boolean markAsUnused;
/**
* Split processing to try and keep each nuke operation relatively short, in
the case that either
* the database or the storage layer is particularly slow.
@@ -117,7 +112,6 @@ public class KillUnusedSegmentsTask extends
AbstractFixedIntervalTask
@JsonProperty("interval") Interval interval,
@JsonProperty("versions") @Nullable List<String> versions,
@JsonProperty("context") Map<String, Object> context,
- @JsonProperty("markAsUnused") @Deprecated Boolean markAsUnused,
@JsonProperty("batchSize") Integer batchSize,
@JsonProperty("limit") @Nullable Integer limit,
@JsonProperty("maxUsedStatusLastUpdatedTime") @Nullable DateTime
maxUsedStatusLastUpdatedTime
@@ -129,7 +123,6 @@ public class KillUnusedSegmentsTask extends
AbstractFixedIntervalTask
interval,
context
);
- this.markAsUnused = markAsUnused != null && markAsUnused;
this.batchSize = (batchSize != null) ? batchSize :
DEFAULT_SEGMENT_NUKE_BATCH_SIZE;
if (this.batchSize <= 0) {
throw InvalidInput.exception("batchSize[%d] must be a positive
integer.", batchSize);
@@ -137,14 +130,6 @@ public class KillUnusedSegmentsTask extends
AbstractFixedIntervalTask
if (limit != null && limit <= 0) {
throw InvalidInput.exception("limit[%d] must be a positive integer.",
limit);
}
- if (Boolean.TRUE.equals(markAsUnused)) {
- if (limit != null) {
- throw InvalidInput.exception("limit[%d] cannot be provided when
markAsUnused is enabled.", limit);
- }
- if (!CollectionUtils.isNullOrEmpty(versions)) {
- throw InvalidInput.exception("versions[%s] cannot be provided when
markAsUnused is enabled.", versions);
- }
- }
this.versions = versions;
this.limit = limit;
this.maxUsedStatusLastUpdatedTime = maxUsedStatusLastUpdatedTime;
@@ -158,21 +143,6 @@ public class KillUnusedSegmentsTask extends
AbstractFixedIntervalTask
return versions;
}
- /**
- * This field has been deprecated as "kill" tasks should not be responsible
for
- * marking segments as unused. Instead, users should call the Coordinator API
- * {@code /{dataSourceName}/markUnused} to explicitly mark segments as
unused.
- * Segments may also be marked unused by the Coordinator if they become
overshadowed
- * or have a {@code DropRule} applied to them.
- */
- @Deprecated
- @JsonProperty
- @JsonInclude(JsonInclude.Include.NON_DEFAULT)
- public boolean isMarkAsUnused()
- {
- return markAsUnused;
- }
-
@JsonProperty
@JsonInclude(JsonInclude.Include.NON_DEFAULT)
public int getBatchSize()
@@ -214,16 +184,6 @@ public class KillUnusedSegmentsTask extends
AbstractFixedIntervalTask
// Track stats for reporting
int numSegmentsKilled = 0;
int numBatchesProcessed = 0;
- final int numSegmentsMarkedAsUnused;
- if (markAsUnused) {
- numSegmentsMarkedAsUnused = toolbox.getTaskActionClient().submit(
- new MarkSegmentsAsUnusedAction(getDataSource(), getInterval())
- );
- LOG.info("Marked [%d] segments of datasource[%s] in interval[%s] as
unused.",
- numSegmentsMarkedAsUnused, getDataSource(), getInterval());
- } else {
- numSegmentsMarkedAsUnused = 0;
- }
// List unused segments
int nextBatchSize = computeNextBatchSize(numSegmentsKilled);
@@ -304,7 +264,7 @@ public class KillUnusedSegmentsTask extends
AbstractFixedIntervalTask
);
final KillTaskReport.Stats stats =
- new KillTaskReport.Stats(numSegmentsKilled, numBatchesProcessed,
numSegmentsMarkedAsUnused);
+ new KillTaskReport.Stats(numSegmentsKilled, numBatchesProcessed);
toolbox.getTaskReportFileWriter().write(
taskId,
TaskReport.buildTaskReports(new KillTaskReport(taskId, stats))
@@ -346,4 +306,41 @@ public class KillUnusedSegmentsTask extends
AbstractFixedIntervalTask
{
return LookupLoadingSpec.NONE;
}
+
+ @Override
+ public boolean isReady(TaskActionClient taskActionClient) throws Exception
+ {
+ final boolean useConcurrentLocks = Boolean.TRUE.equals(
+ getContextValue(
+ Tasks.USE_CONCURRENT_LOCKS,
+ Tasks.DEFAULT_USE_CONCURRENT_LOCKS
+ )
+ );
+
+ TaskLockType actualLockType = determineLockType(useConcurrentLocks);
+
+ final TaskLock lock = taskActionClient.submit(
+ new TimeChunkLockTryAcquireAction(
+ actualLockType,
+ getInterval()
+ )
+ );
+ if (lock == null) {
+ return false;
+ }
+ lock.assertNotRevoked();
+ return true;
+ }
+
+ private TaskLockType determineLockType(boolean useConcurrentLocks)
+ {
+ TaskLockType actualLockType;
+ if (useConcurrentLocks) {
+ actualLockType = TaskLockType.REPLACE;
+ } else {
+ actualLockType = getContextValue(Tasks.TASK_LOCK_TYPE,
TaskLockType.EXCLUSIVE);
+ }
+ return actualLockType;
+ }
+
}
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/ClientKillUnusedSegmentsTaskQuerySerdeTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/ClientKillUnusedSegmentsTaskQuerySerdeTest.java
index e6dbd0bad13..9154a0f5656 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/ClientKillUnusedSegmentsTaskQuerySerdeTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/ClientKillUnusedSegmentsTaskQuerySerdeTest.java
@@ -54,7 +54,6 @@ public class ClientKillUnusedSegmentsTaskQuerySerdeTest
"datasource",
Intervals.of("2020-01-01/P1D"),
null,
- false,
99,
5,
DateTimes.nowUtc()
@@ -65,7 +64,6 @@ public class ClientKillUnusedSegmentsTaskQuerySerdeTest
Assert.assertEquals(taskQuery.getDataSource(), fromJson.getDataSource());
Assert.assertEquals(taskQuery.getInterval(), fromJson.getInterval());
Assert.assertNull(taskQuery.getVersions());
- Assert.assertEquals(taskQuery.getMarkAsUnused(),
fromJson.isMarkAsUnused());
Assert.assertEquals(taskQuery.getBatchSize(),
Integer.valueOf(fromJson.getBatchSize()));
Assert.assertEquals(taskQuery.getLimit(), fromJson.getLimit());
Assert.assertEquals(taskQuery.getMaxUsedStatusLastUpdatedTime(),
fromJson.getMaxUsedStatusLastUpdatedTime());
@@ -79,7 +77,6 @@ public class ClientKillUnusedSegmentsTaskQuerySerdeTest
"datasource",
Intervals.of("2020-01-01/P1D"),
null,
- true,
null,
null,
null
@@ -90,7 +87,6 @@ public class ClientKillUnusedSegmentsTaskQuerySerdeTest
Assert.assertEquals(taskQuery.getDataSource(), fromJson.getDataSource());
Assert.assertEquals(taskQuery.getInterval(), fromJson.getInterval());
Assert.assertNull(taskQuery.getVersions());
- Assert.assertEquals(taskQuery.getMarkAsUnused(),
fromJson.isMarkAsUnused());
Assert.assertEquals(100, fromJson.getBatchSize());
Assert.assertNull(taskQuery.getLimit());
Assert.assertNull(taskQuery.getMaxUsedStatusLastUpdatedTime());
@@ -105,7 +101,6 @@ public class ClientKillUnusedSegmentsTaskQuerySerdeTest
Intervals.of("2020-01-01/P1D"),
null,
null,
- true,
99,
null,
null
@@ -119,7 +114,6 @@ public class ClientKillUnusedSegmentsTaskQuerySerdeTest
Assert.assertEquals(task.getDataSource(), taskQuery.getDataSource());
Assert.assertEquals(task.getInterval(), taskQuery.getInterval());
Assert.assertNull(taskQuery.getVersions());
- Assert.assertEquals(task.isMarkAsUnused(), taskQuery.getMarkAsUnused());
Assert.assertEquals(Integer.valueOf(task.getBatchSize()),
taskQuery.getBatchSize());
Assert.assertNull(taskQuery.getLimit());
Assert.assertNull(taskQuery.getMaxUsedStatusLastUpdatedTime());
@@ -134,7 +128,6 @@ public class ClientKillUnusedSegmentsTaskQuerySerdeTest
Intervals.of("2020-01-01/P1D"),
ImmutableList.of("v1", "v2"),
null,
- null,
99,
100,
DateTimes.nowUtc()
@@ -148,7 +141,6 @@ public class ClientKillUnusedSegmentsTaskQuerySerdeTest
Assert.assertEquals(task.getDataSource(), taskQuery.getDataSource());
Assert.assertEquals(task.getInterval(), taskQuery.getInterval());
Assert.assertEquals(task.getVersions(), taskQuery.getVersions());
- Assert.assertNull(taskQuery.getMarkAsUnused());
Assert.assertEquals(Integer.valueOf(task.getBatchSize()),
taskQuery.getBatchSize());
Assert.assertEquals(task.getLimit(), taskQuery.getLimit());
Assert.assertEquals(task.getMaxUsedStatusLastUpdatedTime(),
taskQuery.getMaxUsedStatusLastUpdatedTime());
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 47c52ad2ba0..f888ace5a54 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
@@ -27,6 +27,10 @@ import org.apache.druid.error.DruidExceptionMatcher;
import org.apache.druid.indexer.TaskState;
import org.apache.druid.indexer.report.KillTaskReport;
import org.apache.druid.indexer.report.TaskReport;
+import org.apache.druid.indexing.common.SegmentLock;
+import org.apache.druid.indexing.common.TaskLockType;
+import org.apache.druid.indexing.common.actions.TaskActionClient;
+import org.apache.druid.indexing.common.actions.TimeChunkLockTryAcquireAction;
import org.apache.druid.indexing.overlord.Segments;
import org.apache.druid.java.util.common.DateTimes;
import org.apache.druid.java.util.common.ISE;
@@ -35,6 +39,8 @@ import org.apache.druid.java.util.common.JodaUtils;
import org.apache.druid.server.lookup.cache.LookupLoadingSpec;
import org.apache.druid.timeline.DataSegment;
import org.assertj.core.api.Assertions;
+import org.easymock.Capture;
+import org.easymock.EasyMock;
import org.hamcrest.MatcherAssert;
import org.joda.time.DateTime;
import org.joda.time.Interval;
@@ -116,52 +122,7 @@ public class KillUnusedSegmentsTaskTest extends
IngestionTestBase
).containsExactlyInAnyOrder(segment1, segment4);
Assert.assertEquals(
- new KillTaskReport.Stats(1, 2, 0),
- getReportedStats()
- );
- }
-
- @Test
- public void testKillWithMarkUnused() throws Exception
- {
- final Set<DataSegment> segments = ImmutableSet.of(segment1, segment2,
segment3, segment4);
- final Set<DataSegment> announced =
getMetadataStorageCoordinator().commitSegments(segments, null);
- Assert.assertEquals(segments, announced);
-
- Assert.assertTrue(
- getSegmentsMetadataManager().markSegmentAsUnused(
- segment2.getId()
- )
- );
-
- final KillUnusedSegmentsTask task = new KillUnusedSegmentsTaskBuilder()
- .dataSource(DATA_SOURCE)
- .interval(Intervals.of("2019-03-01/2019-04-01"))
- .markAsUnused(true)
- .build();
-
- Assert.assertEquals(TaskState.SUCCESS,
taskRunner.run(task).get().getStatusCode());
-
- final List<DataSegment> observedUnusedSegments =
- getMetadataStorageCoordinator().retrieveUnusedSegmentsForInterval(
- DATA_SOURCE,
- Intervals.of("2019/2020"),
- null,
- null,
- null
- );
-
- Assert.assertEquals(ImmutableList.of(segment2), observedUnusedSegments);
- Assertions.assertThat(
- getMetadataStorageCoordinator().retrieveUsedSegmentsForInterval(
- DATA_SOURCE,
- Intervals.of("2019/2020"),
- Segments.ONLY_VISIBLE
- )
- ).containsExactlyInAnyOrder(segment1, segment4);
-
- Assert.assertEquals(
- new KillTaskReport.Stats(1, 2, 1),
+ new KillTaskReport.Stats(1, 2),
getReportedStats()
);
}
@@ -199,7 +160,7 @@ public class KillUnusedSegmentsTaskTest extends
IngestionTestBase
Assert.assertEquals(TaskState.SUCCESS,
taskRunner.run(task).get().getStatusCode());
Assert.assertEquals(
- new KillTaskReport.Stats(4, 3, 0),
+ new KillTaskReport.Stats(4, 3),
getReportedStats()
);
@@ -247,7 +208,7 @@ public class KillUnusedSegmentsTaskTest extends
IngestionTestBase
Assert.assertEquals(TaskState.SUCCESS,
taskRunner.run(task).get().getStatusCode());
Assert.assertEquals(
- new KillTaskReport.Stats(0, 1, 0),
+ new KillTaskReport.Stats(0, 1),
getReportedStats()
);
@@ -296,7 +257,7 @@ public class KillUnusedSegmentsTaskTest extends
IngestionTestBase
Assert.assertEquals(TaskState.SUCCESS,
taskRunner.run(task).get().getStatusCode());
Assert.assertEquals(
- new KillTaskReport.Stats(2, 1, 0),
+ new KillTaskReport.Stats(2, 1),
getReportedStats()
);
@@ -345,7 +306,7 @@ public class KillUnusedSegmentsTaskTest extends
IngestionTestBase
Assert.assertEquals(TaskState.SUCCESS,
taskRunner.run(task).get().getStatusCode());
Assert.assertEquals(
- new KillTaskReport.Stats(0, 1, 0),
+ new KillTaskReport.Stats(0, 1),
getReportedStats()
);
@@ -398,7 +359,7 @@ public class KillUnusedSegmentsTaskTest extends
IngestionTestBase
Assert.assertEquals(TaskState.SUCCESS,
taskRunner.run(task).get().getStatusCode());
Assert.assertEquals(
- new KillTaskReport.Stats(0, 1, 0),
+ new KillTaskReport.Stats(0, 1),
getReportedStats()
);
@@ -419,7 +380,6 @@ public class KillUnusedSegmentsTaskTest extends
IngestionTestBase
final KillUnusedSegmentsTask task = new KillUnusedSegmentsTaskBuilder()
.dataSource(DATA_SOURCE)
.interval(Intervals.of("2019-03-01/2019-04-01"))
- .markAsUnused(true)
.build();
Assert.assertTrue(task.getInputSourceResources().isEmpty());
}
@@ -430,7 +390,6 @@ public class KillUnusedSegmentsTaskTest extends
IngestionTestBase
final KillUnusedSegmentsTask task = new KillUnusedSegmentsTaskBuilder()
.dataSource(DATA_SOURCE)
.interval(Intervals.of("2019-03-01/2019-04-01"))
- .markAsUnused(true)
.build();
Assert.assertEquals(LookupLoadingSpec.Mode.NONE,
task.getLookupLoadingSpec().getMode());
}
@@ -472,7 +431,7 @@ public class KillUnusedSegmentsTaskTest extends
IngestionTestBase
Assert.assertEquals(Collections.emptyList(), observedUnusedSegments);
Assert.assertEquals(
- new KillTaskReport.Stats(4, 4, 0),
+ new KillTaskReport.Stats(4, 4),
getReportedStats()
);
}
@@ -542,7 +501,7 @@ public class KillUnusedSegmentsTaskTest extends
IngestionTestBase
Assert.assertEquals(ImmutableList.of(), observedUnusedSegments);
Assert.assertEquals(
- new KillTaskReport.Stats(3, 4, 0),
+ new KillTaskReport.Stats(3, 4),
getReportedStats()
);
}
@@ -628,7 +587,7 @@ public class KillUnusedSegmentsTaskTest extends
IngestionTestBase
Assert.assertEquals(ImmutableList.of(segment3), observedUnusedSegments);
Assert.assertEquals(
- new KillTaskReport.Stats(2, 3, 0),
+ new KillTaskReport.Stats(2, 3),
getReportedStats()
);
@@ -652,7 +611,7 @@ public class KillUnusedSegmentsTaskTest extends
IngestionTestBase
Assert.assertEquals(ImmutableList.of(), observedUnusedSegments2);
Assert.assertEquals(
- new KillTaskReport.Stats(1, 2, 0),
+ new KillTaskReport.Stats(1, 2),
getReportedStats()
);
}
@@ -732,7 +691,7 @@ public class KillUnusedSegmentsTaskTest extends
IngestionTestBase
Assert.assertEquals(ImmutableList.of(segment2, segment3),
observedUnusedSegments1);
Assert.assertEquals(
- new KillTaskReport.Stats(2, 3, 0),
+ new KillTaskReport.Stats(2, 3),
getReportedStats()
);
@@ -756,7 +715,7 @@ public class KillUnusedSegmentsTaskTest extends
IngestionTestBase
Assert.assertEquals(ImmutableList.of(), observedUnusedSegments2);
Assert.assertEquals(
- new KillTaskReport.Stats(2, 3, 0),
+ new KillTaskReport.Stats(2, 3),
getReportedStats()
);
}
@@ -813,7 +772,7 @@ public class KillUnusedSegmentsTaskTest extends
IngestionTestBase
Assert.assertEquals(TaskState.SUCCESS,
taskRunner.run(task1).get().getStatusCode());
Assert.assertEquals(
- new KillTaskReport.Stats(2, 3, 0),
+ new KillTaskReport.Stats(2, 3),
getReportedStats()
);
@@ -838,7 +797,7 @@ public class KillUnusedSegmentsTaskTest extends
IngestionTestBase
Assert.assertEquals(TaskState.SUCCESS,
taskRunner.run(task2).get().getStatusCode());
Assert.assertEquals(
- new KillTaskReport.Stats(1, 2, 0),
+ new KillTaskReport.Stats(1, 2),
getReportedStats()
);
@@ -861,10 +820,17 @@ public class KillUnusedSegmentsTaskTest extends
IngestionTestBase
Assert.assertEquals(segments, announced);
+ for (DataSegment segment : segments) {
+ Assert.assertTrue(
+ getSegmentsMetadataManager().markSegmentAsUnused(
+ segment.getId()
+ )
+ );
+ }
+
final KillUnusedSegmentsTask task = new KillUnusedSegmentsTaskBuilder()
.dataSource(DATA_SOURCE)
.interval(Intervals.of("2018-01-01/2020-01-01"))
- .markAsUnused(true)
.batchSize(3)
.build();
@@ -880,7 +846,7 @@ public class KillUnusedSegmentsTaskTest extends
IngestionTestBase
Assert.assertEquals(Collections.emptyList(), observedUnusedSegments);
Assert.assertEquals(
- new KillTaskReport.Stats(4, 3, 4),
+ new KillTaskReport.Stats(4, 3),
getReportedStats()
);
}
@@ -989,45 +955,6 @@ public class KillUnusedSegmentsTaskTest extends
IngestionTestBase
);
}
- @Test
- public void testInvalidLimitWithMarkAsUnused()
- {
- MatcherAssert.assertThat(
- Assert.assertThrows(
- DruidException.class,
- () -> new KillUnusedSegmentsTaskBuilder()
- .dataSource(DATA_SOURCE)
- .interval(Intervals.of("2018-01-01/2020-01-01"))
- .markAsUnused(true)
- .batchSize(10)
- .limit(10)
- .build()
- ),
- DruidExceptionMatcher.invalidInput().expectMessageIs(
- "limit[10] cannot be provided when markAsUnused is enabled."
- )
- );
- }
-
- @Test
- public void testInvalidVersionsWithMarkAsUnused()
- {
- MatcherAssert.assertThat(
- Assert.assertThrows(
- DruidException.class,
- () -> new KillUnusedSegmentsTaskBuilder()
- .dataSource(DATA_SOURCE)
- .interval(Intervals.of("2018-01-01/2020-01-01"))
- .markAsUnused(true)
- .versions(ImmutableList.of("foo"))
- .build()
- ),
- DruidExceptionMatcher.invalidInput().expectMessageIs(
- "versions[[foo]] cannot be provided when markAsUnused is enabled."
- )
- );
- }
-
@Test
public void testGetNumTotalBatchesWithBatchSizeSmallerThanLimit()
{
@@ -1046,7 +973,7 @@ public class KillUnusedSegmentsTaskTest extends
IngestionTestBase
{
final String taskId = "test_serde_task";
- final KillTaskReport.Stats stats = new KillTaskReport.Stats(1, 2, 3);
+ final KillTaskReport.Stats stats = new KillTaskReport.Stats(1, 2);
KillTaskReport report = new KillTaskReport(taskId, stats);
String json = getObjectMapper().writeValueAsString(report);
@@ -1059,6 +986,150 @@ public class KillUnusedSegmentsTaskTest extends
IngestionTestBase
Assert.assertEquals(stats, deserializedKillReport.getPayload());
}
+
+ @Test
+ public void testIsReadyWithExclusiveLock() throws Exception
+ {
+ final KillUnusedSegmentsTask task = new KillUnusedSegmentsTaskBuilder()
+ .dataSource(DATA_SOURCE)
+ .interval(Intervals.of("2019-03-01/2019-04-01"))
+ .build();
+
+ Capture<TimeChunkLockTryAcquireAction> acquireActionCapture =
Capture.newInstance();
+
+ TaskActionClient taskActionClient =
EasyMock.createMock(TaskActionClient.class);
+
EasyMock.expect(taskActionClient.submit(EasyMock.capture(acquireActionCapture))).andReturn(
+ new SegmentLock(
+ TaskLockType.EXCLUSIVE,
+ "groupId",
+ "datasource",
+ task.getInterval(),
+ "v1",
+ 0,
+ 0
+ )
+ );
+ EasyMock.replay(taskActionClient);
+
+ Assert.assertTrue(task.isReady(taskActionClient));
+
+ Assert.assertEquals(TaskLockType.EXCLUSIVE,
acquireActionCapture.getValue().getType());
+ }
+
+ @Test
+ public void testIsReadyWithReplaceLock() throws Exception
+ {
+ final KillUnusedSegmentsTask task = new KillUnusedSegmentsTaskBuilder()
+ .dataSource(DATA_SOURCE)
+ .context(ImmutableMap.of(Tasks.USE_CONCURRENT_LOCKS, Boolean.TRUE))
+ .interval(Intervals.of("2019-03-01/2019-04-01"))
+ .build();
+
+ Capture<TimeChunkLockTryAcquireAction> acquireActionCapture =
Capture.newInstance();
+
+ TaskActionClient taskActionClient =
EasyMock.createMock(TaskActionClient.class);
+
EasyMock.expect(taskActionClient.submit(EasyMock.capture(acquireActionCapture)))
+ .andReturn(
+ new SegmentLock(
+ TaskLockType.REPLACE,
+ "groupId",
+ "datasource",
+ task.getInterval(),
+ "v1",
+ 0,
+ 0
+ )
+ );
+ EasyMock.replay(taskActionClient);
+
+ Assert.assertTrue(task.isReady(taskActionClient));
+
+ Assert.assertEquals(TaskLockType.REPLACE,
acquireActionCapture.getValue().getType());
+ }
+
+ @Test
+ public void testIsReadyWithContextAppendLock() throws Exception
+ {
+ final KillUnusedSegmentsTask task = new KillUnusedSegmentsTaskBuilder()
+ .dataSource(DATA_SOURCE)
+ .context(ImmutableMap.of(Tasks.TASK_LOCK_TYPE, TaskLockType.APPEND))
+ .interval(Intervals.of("2019-03-01/2019-04-01"))
+ .build();
+
+ Capture<TimeChunkLockTryAcquireAction> acquireActionCapture =
Capture.newInstance();
+
+ TaskActionClient taskActionClient =
EasyMock.createMock(TaskActionClient.class);
+
EasyMock.expect(taskActionClient.submit(EasyMock.capture(acquireActionCapture)))
+ .andReturn(
+ new SegmentLock(
+ TaskLockType.APPEND,
+ "groupId",
+ "datasource",
+ task.getInterval(),
+ "v1",
+ 0,
+ 0
+ )
+ );
+ EasyMock.replay(taskActionClient);
+
+ Assert.assertTrue(task.isReady(taskActionClient));
+
+ Assert.assertEquals(TaskLockType.APPEND,
acquireActionCapture.getValue().getType());
+ }
+
+ @Test
+ public void testIsReadyWithConcurrentLockHasPrecedenceOverContextLock()
throws Exception
+ {
+ final KillUnusedSegmentsTask task = new KillUnusedSegmentsTaskBuilder()
+ .dataSource(DATA_SOURCE)
+ .context(ImmutableMap.of(Tasks.USE_CONCURRENT_LOCKS, Boolean.TRUE,
Tasks.TASK_LOCK_TYPE, TaskLockType.APPEND))
+ .interval(Intervals.of("2019-03-01/2019-04-01"))
+ .build();
+
+ Capture<TimeChunkLockTryAcquireAction> acquireActionCapture =
Capture.newInstance();
+
+ TaskActionClient taskActionClient =
EasyMock.createMock(TaskActionClient.class);
+
EasyMock.expect(taskActionClient.submit(EasyMock.capture(acquireActionCapture)))
+ .andReturn(
+ new SegmentLock(
+ TaskLockType.REPLACE,
+ "groupId",
+ "datasource",
+ task.getInterval(),
+ "v1",
+ 0,
+ 0
+ )
+ );
+ EasyMock.replay(taskActionClient);
+
+ Assert.assertTrue(task.isReady(taskActionClient));
+
+ Assert.assertEquals(TaskLockType.REPLACE,
acquireActionCapture.getValue().getType());
+ }
+
+
+ @Test
+ public void testIsReadyReturnsNullLock() throws Exception
+ {
+ final KillUnusedSegmentsTask task = new KillUnusedSegmentsTaskBuilder()
+ .dataSource(DATA_SOURCE)
+ .context(ImmutableMap.of(Tasks.USE_CONCURRENT_LOCKS, Boolean.TRUE))
+ .interval(Intervals.of("2019-03-01/2019-04-01"))
+ .build();
+
+ Capture<TimeChunkLockTryAcquireAction> acquireActionCapture =
Capture.newInstance();
+
+ TaskActionClient taskActionClient =
EasyMock.createMock(TaskActionClient.class);
+
EasyMock.expect(taskActionClient.submit(EasyMock.capture(acquireActionCapture))).andReturn(null);
+ EasyMock.replay(taskActionClient);
+
+ Assert.assertFalse(task.isReady(taskActionClient));
+
+ Assert.assertEquals(TaskLockType.REPLACE,
acquireActionCapture.getValue().getType());
+ }
+
private static class KillUnusedSegmentsTaskBuilder
{
private String id;
@@ -1066,7 +1137,6 @@ public class KillUnusedSegmentsTaskTest extends
IngestionTestBase
private Interval interval;
private List<String> versions;
private Map<String, Object> context;
- private Boolean markAsUnused;
private Integer batchSize;
private Integer limit;
private DateTime maxUsedStatusLastUpdatedTime;
@@ -1101,12 +1171,6 @@ public class KillUnusedSegmentsTaskTest extends
IngestionTestBase
return this;
}
- public KillUnusedSegmentsTaskBuilder markAsUnused(Boolean markAsUnused)
- {
- this.markAsUnused = markAsUnused;
- return this;
- }
-
public KillUnusedSegmentsTaskBuilder batchSize(Integer batchSize)
{
this.batchSize = batchSize;
@@ -1133,7 +1197,6 @@ public class KillUnusedSegmentsTaskTest extends
IngestionTestBase
interval,
versions,
context,
- markAsUnused,
batchSize,
limit,
maxUsedStatusLastUpdatedTime
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/PushedSegmentsReportTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/PushedSegmentsReportTest.java
index b46f3da5b1f..f42f7d4b27b 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/PushedSegmentsReportTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/PushedSegmentsReportTest.java
@@ -31,7 +31,7 @@ public class PushedSegmentsReportTest
{
TaskReport.ReportMap map1 = new TaskReport.ReportMap();
TaskReport.ReportMap map2 = new TaskReport.ReportMap();
- map2.put("killTaskReport", new KillTaskReport("taskId", new
KillTaskReport.Stats(1, 2, 3)));
+ map2.put("killTaskReport", new KillTaskReport("taskId", new
KillTaskReport.Stats(1, 2)));
EqualsVerifier.forClass(PushedSegmentsReport.class)
.usingGetClass()
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskLifecycleTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskLifecycleTest.java
index ece18aa852d..e94f593ccb9 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskLifecycleTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskLifecycleTest.java
@@ -939,7 +939,6 @@ public class TaskLifecycleTest extends
InitializedNullHandlingTest
Intervals.of("2011-04-01/P4D"),
null,
null,
- false,
null,
null,
null
@@ -1038,7 +1037,6 @@ public class TaskLifecycleTest extends
InitializedNullHandlingTest
Intervals.of("2011-04-01/P4D"),
null,
null,
- false,
null,
maxSegmentsToKill,
null
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/http/OverlordResourceTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/http/OverlordResourceTest.java
index 41626341bae..51b0cfe742e 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/http/OverlordResourceTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/http/OverlordResourceTest.java
@@ -926,7 +926,7 @@ public class OverlordResourceTest
auditManager
);
- Task task = new KillUnusedSegmentsTask("kill_all", "allow",
Intervals.ETERNITY, null, null, false, 10, null, null);
+ Task task = new KillUnusedSegmentsTask("kill_all", "allow",
Intervals.ETERNITY, null, null, 10, null, null);
overlordResource.taskPost(task, req);
Assert.assertTrue(auditEntryCapture.hasCaptured());
diff --git
a/processing/src/main/java/org/apache/druid/indexer/report/KillTaskReport.java
b/processing/src/main/java/org/apache/druid/indexer/report/KillTaskReport.java
index c0736a9e1a9..a877d2bf7b1 100644
---
a/processing/src/main/java/org/apache/druid/indexer/report/KillTaskReport.java
+++
b/processing/src/main/java/org/apache/druid/indexer/report/KillTaskReport.java
@@ -22,7 +22,6 @@ package org.apache.druid.indexer.report;
import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
-import javax.annotation.Nullable;
import java.util.Objects;
public class KillTaskReport implements TaskReport
@@ -85,18 +84,15 @@ public class KillTaskReport implements TaskReport
{
private final int numSegmentsKilled;
private final int numBatchesProcessed;
- private final Integer numSegmentsMarkedAsUnused;
@JsonCreator
public Stats(
@JsonProperty("numSegmentsKilled") int numSegmentsKilled,
- @JsonProperty("numBatchesProcessed") int numBatchesProcessed,
- @JsonProperty("numSegmentsMarkedAsUnused") @Nullable Integer
numSegmentsMarkedAsUnused
+ @JsonProperty("numBatchesProcessed") int numBatchesProcessed
)
{
this.numSegmentsKilled = numSegmentsKilled;
this.numBatchesProcessed = numBatchesProcessed;
- this.numSegmentsMarkedAsUnused = numSegmentsMarkedAsUnused;
}
@JsonProperty
@@ -111,13 +107,6 @@ public class KillTaskReport implements TaskReport
return numBatchesProcessed;
}
- @Nullable
- @JsonProperty
- public Integer getNumSegmentsMarkedAsUnused()
- {
- return numSegmentsMarkedAsUnused;
- }
-
@Override
public boolean equals(Object o)
{
@@ -129,14 +118,13 @@ public class KillTaskReport implements TaskReport
}
Stats that = (Stats) o;
return numSegmentsKilled == that.numSegmentsKilled
- && numBatchesProcessed == that.numBatchesProcessed
- && Objects.equals(this.numSegmentsMarkedAsUnused,
that.numSegmentsMarkedAsUnused);
+ && numBatchesProcessed == that.numBatchesProcessed;
}
@Override
public int hashCode()
{
- return Objects.hash(numSegmentsKilled, numBatchesProcessed,
numSegmentsMarkedAsUnused);
+ return Objects.hash(numSegmentsKilled, numBatchesProcessed);
}
@Override
@@ -145,7 +133,6 @@ public class KillTaskReport implements TaskReport
return "Stats{" +
"numSegmentsKilled=" + numSegmentsKilled +
", numBatchesProcessed=" + numBatchesProcessed +
- ", numSegmentsMarkedAsUnused=" + numSegmentsMarkedAsUnused +
'}';
}
}
diff --git
a/processing/src/test/java/org/apache/druid/indexer/report/TaskReportSerdeTest.java
b/processing/src/test/java/org/apache/druid/indexer/report/TaskReportSerdeTest.java
index d71964eed66..4ff36d731df 100644
---
a/processing/src/test/java/org/apache/druid/indexer/report/TaskReportSerdeTest.java
+++
b/processing/src/test/java/org/apache/druid/indexer/report/TaskReportSerdeTest.java
@@ -73,7 +73,7 @@ public class TaskReportSerdeTest
@Test
public void testSerdeOfKillTaskReport() throws Exception
{
- KillTaskReport originalReport = new KillTaskReport("taskId", new
KillTaskReport.Stats(1, 2, 3));
+ KillTaskReport originalReport = new KillTaskReport("taskId", new
KillTaskReport.Stats(1, 2));
String reportJson = jsonMapper.writeValueAsString(originalReport);
TaskReport deserialized = jsonMapper.readValue(reportJson,
TaskReport.class);
@@ -81,6 +81,7 @@ public class TaskReportSerdeTest
KillTaskReport deserializedReport = (KillTaskReport) deserialized;
Assert.assertEquals(originalReport, deserializedReport);
+ Assert.assertEquals(originalReport.hashCode(),
deserializedReport.hashCode());
}
@Test
diff --git
a/server/src/main/java/org/apache/druid/client/indexing/ClientKillUnusedSegmentsTaskQuery.java
b/server/src/main/java/org/apache/druid/client/indexing/ClientKillUnusedSegmentsTaskQuery.java
index 4dfad3c97c0..650a79a75e7 100644
---
a/server/src/main/java/org/apache/druid/client/indexing/ClientKillUnusedSegmentsTaskQuery.java
+++
b/server/src/main/java/org/apache/druid/client/indexing/ClientKillUnusedSegmentsTaskQuery.java
@@ -43,7 +43,6 @@ public class ClientKillUnusedSegmentsTaskQuery implements
ClientTaskQuery
private final Interval interval;
@Nullable
private final List<String> versions;
- private final Boolean markAsUnused;
private final Integer batchSize;
@Nullable
private final Integer limit;
@@ -56,7 +55,6 @@ public class ClientKillUnusedSegmentsTaskQuery implements
ClientTaskQuery
@JsonProperty("dataSource") String dataSource,
@JsonProperty("interval") Interval interval,
@JsonProperty("versions") @Nullable List<String> versions,
- @JsonProperty("markAsUnused") @Deprecated Boolean markAsUnused,
@JsonProperty("batchSize") Integer batchSize,
@JsonProperty("limit") @Nullable Integer limit,
@JsonProperty("maxUsedStatusLastUpdatedTime") @Nullable DateTime
maxUsedStatusLastUpdatedTime
@@ -72,7 +70,6 @@ public class ClientKillUnusedSegmentsTaskQuery implements
ClientTaskQuery
this.dataSource = dataSource;
this.interval = interval;
this.versions = versions;
- this.markAsUnused = markAsUnused;
this.batchSize = batchSize;
this.limit = limit;
this.maxUsedStatusLastUpdatedTime = maxUsedStatusLastUpdatedTime;
@@ -112,20 +109,6 @@ public class ClientKillUnusedSegmentsTaskQuery implements
ClientTaskQuery
return versions;
}
- /**
- * This field has been deprecated as "kill" tasks should not be responsible
for
- * marking segments as unused. Instead, users should call the Coordinator API
- * {@code /{dataSourceName}/markUnused} to explicitly mark segments as
unused.
- * Segments may also be marked unused by the Coordinator if they become
overshadowed
- * or have a {@code DropRule} applied to them.
- */
- @Deprecated
- @JsonProperty
- public Boolean getMarkAsUnused()
- {
- return markAsUnused;
- }
-
@JsonProperty
public Integer getBatchSize()
{
@@ -161,7 +144,6 @@ public class ClientKillUnusedSegmentsTaskQuery implements
ClientTaskQuery
&& Objects.equals(dataSource, that.dataSource)
&& Objects.equals(interval, that.interval)
&& Objects.equals(versions, that.versions)
- && Objects.equals(markAsUnused, that.markAsUnused)
&& Objects.equals(batchSize, that.batchSize)
&& Objects.equals(limit, that.limit)
&& Objects.equals(maxUsedStatusLastUpdatedTime,
that.maxUsedStatusLastUpdatedTime);
@@ -170,6 +152,6 @@ public class ClientKillUnusedSegmentsTaskQuery implements
ClientTaskQuery
@Override
public int hashCode()
{
- return Objects.hash(id, dataSource, interval, versions, markAsUnused,
batchSize, limit, maxUsedStatusLastUpdatedTime);
+ return Objects.hash(id, dataSource, interval, versions, batchSize, limit,
maxUsedStatusLastUpdatedTime);
}
}
diff --git
a/server/src/main/java/org/apache/druid/rpc/indexing/OverlordClient.java
b/server/src/main/java/org/apache/druid/rpc/indexing/OverlordClient.java
index 2d4dfe4aae7..310684206d2 100644
--- a/server/src/main/java/org/apache/druid/rpc/indexing/OverlordClient.java
+++ b/server/src/main/java/org/apache/druid/rpc/indexing/OverlordClient.java
@@ -109,7 +109,6 @@ public interface OverlordClient
dataSource,
interval,
versions,
- false,
null,
maxSegmentsToKill,
maxUsedStatusLastUpdatedTime
diff --git
a/server/src/test/java/org/apache/druid/client/indexing/ClientKillUnusedSegmentsTaskQueryTest.java
b/server/src/test/java/org/apache/druid/client/indexing/ClientKillUnusedSegmentsTaskQueryTest.java
index d683d59ac6e..9f9eb8075ef 100644
---
a/server/src/test/java/org/apache/druid/client/indexing/ClientKillUnusedSegmentsTaskQueryTest.java
+++
b/server/src/test/java/org/apache/druid/client/indexing/ClientKillUnusedSegmentsTaskQueryTest.java
@@ -33,7 +33,6 @@ public class ClientKillUnusedSegmentsTaskQueryTest
private static final String DATA_SOURCE = "data_source";
public static final DateTime START = DateTimes.nowUtc();
private static final Interval INTERVAL = new Interval(START, START.plus(1));
- private static final Boolean MARK_UNUSED = true;
private static final Integer BATCH_SIZE = 999;
private static final Integer LIMIT = 1000;
@@ -47,7 +46,6 @@ public class ClientKillUnusedSegmentsTaskQueryTest
DATA_SOURCE,
INTERVAL,
null,
- true,
BATCH_SIZE,
LIMIT,
null
@@ -78,12 +76,6 @@ public class ClientKillUnusedSegmentsTaskQueryTest
Assert.assertEquals(INTERVAL, clientKillUnusedSegmentsQuery.getInterval());
}
- @Test
- public void testGetMarkUnused()
- {
- Assert.assertEquals(MARK_UNUSED,
clientKillUnusedSegmentsQuery.getMarkAsUnused());
- }
-
@Test
public void testGetBatchSize()
{
diff --git
a/server/src/test/java/org/apache/druid/rpc/indexing/OverlordClientImplTest.java
b/server/src/test/java/org/apache/druid/rpc/indexing/OverlordClientImplTest.java
index 57bab1fed0d..8c3b867e368 100644
---
a/server/src/test/java/org/apache/druid/rpc/indexing/OverlordClientImplTest.java
+++
b/server/src/test/java/org/apache/druid/rpc/indexing/OverlordClientImplTest.java
@@ -436,7 +436,6 @@ public class OverlordClientImplTest
null,
null,
null,
- null,
null
);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
