You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by kf...@apache.org on 2024/01/15 04:39:07 UTC
(druid) branch master updated: Handle and map errors in delete pending segments API (#15673)
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 08c01f1dae9 Handle and map errors in delete pending segments API (#15673)
08c01f1dae9 is described below
commit 08c01f1dae9732859ee69f12b4612c6df05c4dde
Author: Abhishek Radhakrishnan <ab...@gmail.com>
AuthorDate: Sun Jan 14 20:39:01 2024 -0800
Handle and map errors in delete pending segments API (#15673)
Changes:
- Handle exception in deletePendingSegments API and map to correct HTTP status code
- Clean up exception message using `DruidException`
- Add unit tests
---
.../overlord/IndexerMetadataStorageAdapter.java | 39 +++---
.../apache/druid/indexing/overlord/TaskMaster.java | 2 +-
.../indexing/overlord/http/OverlordResource.java | 21 +++-
.../IndexerMetadataStorageAdapterTest.java | 75 ++++++++++--
.../overlord/http/OverlordResourceTest.java | 136 ++++++++++++++++++++-
5 files changed, 238 insertions(+), 35 deletions(-)
diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/IndexerMetadataStorageAdapter.java b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/IndexerMetadataStorageAdapter.java
index 4671c5bb7a9..6d23fcec327 100644
--- a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/IndexerMetadataStorageAdapter.java
+++ b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/IndexerMetadataStorageAdapter.java
@@ -19,11 +19,12 @@
package org.apache.druid.indexing.overlord;
-import com.google.common.base.Preconditions;
import com.google.inject.Inject;
+import org.apache.druid.error.InvalidInput;
import org.apache.druid.indexer.TaskInfo;
+import org.apache.druid.indexer.TaskStatus;
+import org.apache.druid.indexing.common.task.Task;
import org.apache.druid.java.util.common.DateTimes;
-import org.joda.time.DateTime;
import org.joda.time.Interval;
import java.util.Comparator;
@@ -46,23 +47,31 @@ public class IndexerMetadataStorageAdapter
public int deletePendingSegments(String dataSource, Interval deleteInterval)
{
- // Check the given interval overlaps the interval(minCreatedDateOfActiveTasks, MAX)
- final Optional<DateTime> minCreatedDateOfActiveTasks = taskStorageQueryAdapter
+ // Find the earliest active task created for the specified datasource; if one exists,
+ // check if its interval overlaps with the delete interval.
+ final Optional<TaskInfo<Task, TaskStatus>> earliestActiveTaskOptional = taskStorageQueryAdapter
.getActiveTaskInfo(dataSource)
.stream()
- .map(TaskInfo::getCreatedTime)
- .min(Comparator.naturalOrder());
+ .min(Comparator.comparing(TaskInfo::getCreatedTime));
- final Interval activeTaskInterval = new Interval(
- minCreatedDateOfActiveTasks.orElse(DateTimes.MAX),
- DateTimes.MAX
- );
+ if (earliestActiveTaskOptional.isPresent()) {
+ final TaskInfo<Task, TaskStatus> earliestActiveTask = earliestActiveTaskOptional.get();
+ final Interval activeTaskInterval = new Interval(
+ earliestActiveTask.getCreatedTime(),
+ DateTimes.MAX
+ );
- Preconditions.checkArgument(
- !deleteInterval.overlaps(activeTaskInterval),
- "Cannot delete pendingSegments because there is at least one active task created at %s",
- activeTaskInterval.getStart()
- );
+ if (deleteInterval.overlaps(activeTaskInterval)) {
+ throw InvalidInput.exception(
+ "Cannot delete pendingSegments for datasource[%s] as there is at least one active task[%s] created at[%s] "
+ + "that overlaps with the delete interval[%s]. Please retry when there are no active tasks.",
+ dataSource,
+ earliestActiveTask.getId(),
+ activeTaskInterval.getStart(),
+ deleteInterval
+ );
+ }
+ }
return indexerMetadataStorageCoordinator.deletePendingSegmentsCreatedInInterval(dataSource, deleteInterval);
}
diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskMaster.java b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskMaster.java
index e4d26b37573..4798513aafd 100644
--- a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskMaster.java
+++ b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskMaster.java
@@ -238,7 +238,7 @@ public class TaskMaster implements TaskCountStatsProvider, TaskSlotCountStatsPro
}
/**
- * Returns true if it's the leader and its all services have been properly initialized.
+ * Returns true if it's the leader and all its services have been initialized.
*/
public boolean isLeader()
{
diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/http/OverlordResource.java b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/http/OverlordResource.java
index 7afaaa90526..b589dc5ffd8 100644
--- a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/http/OverlordResource.java
+++ b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/http/OverlordResource.java
@@ -939,10 +939,25 @@ public class OverlordResource
}
if (taskMaster.isLeader()) {
- final int numDeleted = indexerMetadataStorageAdapter.deletePendingSegments(dataSource, deleteInterval);
- return Response.ok().entity(ImmutableMap.of("numDeleted", numDeleted)).build();
+ try {
+ final int numDeleted = indexerMetadataStorageAdapter.deletePendingSegments(dataSource, deleteInterval);
+ return Response.ok().entity(ImmutableMap.of("numDeleted", numDeleted)).build();
+ }
+ catch (DruidException e) {
+ return Response.status(e.getStatusCode())
+ .entity(ImmutableMap.<String, Object>of("error", e.getMessage()))
+ .build();
+ }
+ catch (Exception e) {
+ log.warn(e, "Failed to delete pending segments for datasource[%s] and interval[%s].", dataSource, deleteInterval);
+ return Response.status(Status.INTERNAL_SERVER_ERROR)
+ .entity(ImmutableMap.<String, Object>of("error", e.getMessage()))
+ .build();
+ }
} else {
- return Response.status(Status.SERVICE_UNAVAILABLE).build();
+ return Response.status(Status.SERVICE_UNAVAILABLE)
+ .entity(ImmutableMap.of("error", "overlord is not the leader or not initialized yet"))
+ .build();
}
}
diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/IndexerMetadataStorageAdapterTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/IndexerMetadataStorageAdapterTest.java
index 19605634117..633d861410d 100644
--- a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/IndexerMetadataStorageAdapterTest.java
+++ b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/IndexerMetadataStorageAdapterTest.java
@@ -20,6 +20,8 @@
package org.apache.druid.indexing.overlord;
import com.google.common.collect.ImmutableList;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.error.DruidExceptionMatcher;
import org.apache.druid.indexer.TaskInfo;
import org.apache.druid.indexer.TaskStatus;
import org.apache.druid.indexing.common.task.NoopTask;
@@ -27,21 +29,16 @@ import org.apache.druid.indexing.common.task.Task;
import org.apache.druid.java.util.common.DateTimes;
import org.apache.druid.java.util.common.Intervals;
import org.easymock.EasyMock;
-import org.hamcrest.CoreMatchers;
+import org.hamcrest.MatcherAssert;
import org.joda.time.Interval;
import org.junit.Assert;
import org.junit.Before;
-import org.junit.Rule;
import org.junit.Test;
-import org.junit.rules.ExpectedException;
import java.util.List;
public class IndexerMetadataStorageAdapterTest
{
- @Rule
- public ExpectedException expectedException = ExpectedException.none();
-
private TaskStorageQueryAdapter taskStorageQueryAdapter;
private IndexerMetadataStorageCoordinator indexerMetadataStorageCoordinator;
private IndexerMetadataStorageAdapter indexerMetadataStorageAdapter;
@@ -69,7 +66,7 @@ public class IndexerMetadataStorageAdapterTest
NoopTask.create()
),
new TaskInfo<>(
- "id1",
+ "id2",
DateTimes.of("2017-12-02"),
TaskStatus.running("id2"),
"dataSource",
@@ -93,7 +90,7 @@ public class IndexerMetadataStorageAdapterTest
}
@Test
- public void testDeletePendingSegmentsOfRunningTasks()
+ public void testDeletePendingSegmentsOfOneOverlappingRunningTask()
{
final ImmutableList<TaskInfo<Task, TaskStatus>> taskInfos = ImmutableList.of(
new TaskInfo<>(
@@ -104,7 +101,7 @@ public class IndexerMetadataStorageAdapterTest
NoopTask.create()
),
new TaskInfo<>(
- "id1",
+ "id2",
DateTimes.of("2017-12-02"),
TaskStatus.running("id2"),
"dataSource",
@@ -125,8 +122,62 @@ public class IndexerMetadataStorageAdapterTest
.andReturn(10);
EasyMock.replay(taskStorageQueryAdapter, indexerMetadataStorageCoordinator);
- expectedException.expect(CoreMatchers.instanceOf(IllegalArgumentException.class));
- expectedException.expectMessage("Cannot delete pendingSegments because there is at least one active task created");
- indexerMetadataStorageAdapter.deletePendingSegments("dataSource", deleteInterval);
+ MatcherAssert.assertThat(
+ Assert.assertThrows(
+ DruidException.class,
+ () -> indexerMetadataStorageAdapter.deletePendingSegments("dataSource", deleteInterval)
+ ),
+ DruidExceptionMatcher.invalidInput().expectMessageIs(
+ "Cannot delete pendingSegments for datasource[dataSource] as there is at least one active task[id1]"
+ + " created at[2017-11-01T00:00:00.000Z] that overlaps with the delete "
+ + "interval[2017-01-01T00:00:00.000Z/2017-12-01T00:00:00.000Z]. Please retry when there are no active tasks."
+ )
+ );
+ }
+
+ @Test
+ public void testDeletePendingSegmentsOfMultipleOverlappingRunningTasks()
+ {
+ final ImmutableList<TaskInfo<Task, TaskStatus>> taskInfos = ImmutableList.of(
+ new TaskInfo<>(
+ "id1",
+ DateTimes.of("2017-12-01"),
+ TaskStatus.running("id1"),
+ "dataSource",
+ NoopTask.create()
+ ),
+ new TaskInfo<>(
+ "id2",
+ DateTimes.of("2017-11-01"),
+ TaskStatus.running("id2"),
+ "dataSource",
+ NoopTask.create()
+ )
+ );
+
+ EasyMock.expect(taskStorageQueryAdapter.getActiveTaskInfo("dataSource")).andReturn(taskInfos);
+
+ final Interval deleteInterval = Intervals.of("2017-01-01/2018-12-01");
+ EasyMock
+ .expect(
+ indexerMetadataStorageCoordinator.deletePendingSegmentsCreatedInInterval(
+ EasyMock.anyString(),
+ EasyMock.eq(deleteInterval)
+ )
+ )
+ .andReturn(10);
+ EasyMock.replay(taskStorageQueryAdapter, indexerMetadataStorageCoordinator);
+
+ MatcherAssert.assertThat(
+ Assert.assertThrows(
+ DruidException.class,
+ () -> indexerMetadataStorageAdapter.deletePendingSegments("dataSource", deleteInterval)
+ ),
+ DruidExceptionMatcher.invalidInput().expectMessageIs(
+ "Cannot delete pendingSegments for datasource[dataSource] as there is at least one active task[id2]"
+ + " created at[2017-11-01T00:00:00.000Z] that overlaps with the delete"
+ + " interval[2017-01-01T00:00:00.000Z/2018-12-01T00:00:00.000Z]. Please retry when there are no active tasks."
+ )
+ );
}
}
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 561e8dd9b19..cdaf5b9b359 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
@@ -28,6 +28,8 @@ import com.google.common.collect.ImmutableSet;
import org.apache.druid.audit.AuditEntry;
import org.apache.druid.audit.AuditManager;
import org.apache.druid.common.config.JacksonConfigManager;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.error.InvalidInput;
import org.apache.druid.indexer.RunnerTaskState;
import org.apache.druid.indexer.TaskInfo;
import org.apache.druid.indexer.TaskLocation;
@@ -983,10 +985,136 @@ public class OverlordResourceTest
authConfig
);
- final Map<String, Integer> response = (Map<String, Integer>) overlordResource
- .killPendingSegments("allow", new Interval(DateTimes.MIN, DateTimes.nowUtc()).toString(), req)
- .getEntity();
- Assert.assertEquals(2, response.get("numDeleted").intValue());
+ Response response = overlordResource
+ .killPendingSegments("allow", new Interval(DateTimes.MIN, DateTimes.nowUtc()).toString(), req);
+ Assert.assertEquals(200, response.getStatus());
+ Assert.assertEquals(ImmutableMap.of("numDeleted", 2), response.getEntity());
+ }
+
+ @Test
+ public void testKillPendingSegmentsThrowsInvalidInputDruidException()
+ {
+ expectAuthorizationTokenCheck();
+
+ EasyMock.expect(taskMaster.isLeader()).andReturn(true);
+ final String exceptionMsg = "Some exception msg";
+ EasyMock
+ .expect(
+ indexerMetadataStorageAdapter.deletePendingSegments(
+ EasyMock.eq("allow"),
+ EasyMock.anyObject(Interval.class)
+ )
+ )
+ .andThrow(InvalidInput.exception(exceptionMsg))
+ .once();
+
+ EasyMock.replay(
+ taskRunner,
+ taskMaster,
+ taskStorageQueryAdapter,
+ indexerMetadataStorageAdapter,
+ req,
+ workerTaskRunnerQueryAdapter,
+ authConfig
+ );
+
+ Response response = overlordResource
+ .killPendingSegments("allow", new Interval(DateTimes.MIN, DateTimes.nowUtc()).toString(), req);
+
+ Assert.assertEquals(400, response.getStatus());
+ Assert.assertEquals(ImmutableMap.of("error", exceptionMsg), response.getEntity());
+ }
+
+ @Test
+ public void testKillPendingSegmentsThrowsDefensiveDruidException()
+ {
+ expectAuthorizationTokenCheck();
+
+ EasyMock.expect(taskMaster.isLeader()).andReturn(true);
+ final String exceptionMsg = "An internal defensive exception";
+ EasyMock
+ .expect(
+ indexerMetadataStorageAdapter.deletePendingSegments(
+ EasyMock.eq("allow"),
+ EasyMock.anyObject(Interval.class)
+ )
+ )
+ .andThrow(DruidException.defensive(exceptionMsg))
+ .once();
+
+ EasyMock.replay(
+ taskRunner,
+ taskMaster,
+ taskStorageQueryAdapter,
+ indexerMetadataStorageAdapter,
+ req,
+ workerTaskRunnerQueryAdapter,
+ authConfig
+ );
+
+ Response response = overlordResource
+ .killPendingSegments("allow", new Interval(DateTimes.MIN, DateTimes.nowUtc()).toString(), req);
+
+ Assert.assertEquals(500, response.getStatus());
+ Assert.assertEquals(ImmutableMap.of("error", exceptionMsg), response.getEntity());
+ }
+
+ @Test
+ public void testKillPendingSegmentsThrowsArbitraryException()
+ {
+ expectAuthorizationTokenCheck();
+
+ EasyMock.expect(taskMaster.isLeader()).andReturn(true);
+ final String exceptionMsg = "An unexpected illegal state exception";
+ EasyMock
+ .expect(
+ indexerMetadataStorageAdapter.deletePendingSegments(
+ EasyMock.eq("allow"),
+ EasyMock.anyObject(Interval.class)
+ )
+ )
+ .andThrow(new IllegalStateException(exceptionMsg))
+ .once();
+
+ EasyMock.replay(
+ taskRunner,
+ taskMaster,
+ taskStorageQueryAdapter,
+ indexerMetadataStorageAdapter,
+ req,
+ workerTaskRunnerQueryAdapter,
+ authConfig
+ );
+
+ Response response = overlordResource
+ .killPendingSegments("allow", new Interval(DateTimes.MIN, DateTimes.nowUtc()).toString(), req);
+
+ Assert.assertEquals(500, response.getStatus());
+ Assert.assertEquals(ImmutableMap.of("error", exceptionMsg), response.getEntity());
+ }
+
+ @Test
+ public void testKillPendingSegmentsToNonLeader()
+ {
+ expectAuthorizationTokenCheck();
+
+ EasyMock.expect(taskMaster.isLeader()).andReturn(false);
+
+ EasyMock.replay(
+ taskRunner,
+ taskMaster,
+ taskStorageQueryAdapter,
+ indexerMetadataStorageAdapter,
+ req,
+ workerTaskRunnerQueryAdapter,
+ authConfig
+ );
+
+ Response response = overlordResource
+ .killPendingSegments("allow", new Interval(DateTimes.MIN, DateTimes.nowUtc()).toString(), req);
+
+ Assert.assertEquals(503, response.getStatus());
+ Assert.assertEquals(ImmutableMap.of("error", "overlord is not the leader or not initialized yet"), response.getEntity());
}
@Test
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org