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 d083f6e9b99 minor: Add `LeaderOverlordService` and other miscellaneous
things (#19949)
d083f6e9b99 is described below
commit d083f6e9b9927936e2340d4397a91b8fddb14ed2
Author: Kashif Faraz <[email protected]>
AuthorDate: Tue Aug 11 12:16:44 2026 +0530
minor: Add `LeaderOverlordService` and other miscellaneous things (#19949)
Changes
---------
- Add method `SupervisorManager.getSupervisorOfType`
- Add `LeaderOverlordService` to tie a lifecycle of a stateful service
running
on the Overlord to the leadership lifecycle
- Add `DartWorkerModule.setProperties` to avoid injection of a private field
---
.../indexing/compact/CompactionScheduler.java | 7 +--
.../druid/indexing/overlord/DruidOverlord.java | 12 ++---
.../overlord/supervisor/SupervisorManager.java | 45 +++++++++++------
.../scheduledbatch/ScheduledBatchTaskManager.java | 15 +++---
.../druid/indexing/overlord/http/OverlordTest.java | 7 ++-
.../overlord/supervisor/SupervisorManagerTest.java | 5 +-
.../ScheduledBatchTaskManagerTest.java | 20 ++++----
.../druid/msq/dart/guice/DartWorkerModule.java | 7 ++-
.../indexing/overlord/LeaderOverlordService.java | 58 ++++++++++++++++++++++
.../java/org/apache/druid/cli/CliOverlord.java | 7 +++
.../testing/embedded/EmbeddedServiceClient.java | 7 +--
11 files changed, 137 insertions(+), 53 deletions(-)
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/compact/CompactionScheduler.java
b/indexing-service/src/main/java/org/apache/druid/indexing/compact/CompactionScheduler.java
index 6f5ed1a7a6e..57453540679 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/compact/CompactionScheduler.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/compact/CompactionScheduler.java
@@ -19,6 +19,7 @@
package org.apache.druid.indexing.compact;
+import org.apache.druid.indexing.overlord.LeaderOverlordService;
import org.apache.druid.server.compaction.CompactionSimulateResult;
import org.apache.druid.server.coordinator.AutoCompactionSnapshot;
import org.apache.druid.server.coordinator.ClusterCompactionConfig;
@@ -42,12 +43,8 @@ import java.util.Map;
* should call {@link #stopCompaction}.</li>
* </ul>
*/
-public interface CompactionScheduler
+public interface CompactionScheduler extends LeaderOverlordService
{
- void becomeLeader();
-
- void stopBeingLeader();
-
/**
* @return true if the scheduler is enabled i.e. when
* {@link DruidCompactionConfig#isUseSupervisors()} is true.
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/DruidOverlord.java
b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/DruidOverlord.java
index 4dac9ea81c1..6e7d25168a7 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/DruidOverlord.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/DruidOverlord.java
@@ -27,13 +27,11 @@ import org.apache.druid.discovery.DruidLeaderSelector;
import org.apache.druid.indexing.common.actions.SegmentAllocationQueue;
import org.apache.druid.indexing.common.actions.TaskActionClientFactory;
import org.apache.druid.indexing.common.task.TaskContextEnricher;
-import org.apache.druid.indexing.compact.CompactionScheduler;
import org.apache.druid.indexing.overlord.config.DefaultTaskConfig;
import org.apache.druid.indexing.overlord.config.TaskLockConfig;
import org.apache.druid.indexing.overlord.config.TaskQueueConfig;
import org.apache.druid.indexing.overlord.duty.OverlordDutyExecutor;
import org.apache.druid.indexing.overlord.supervisor.SupervisorManager;
-import org.apache.druid.indexing.scheduledbatch.ScheduledBatchTaskManager;
import org.apache.druid.java.util.common.lifecycle.Lifecycle;
import org.apache.druid.java.util.common.lifecycle.LifecycleStart;
import org.apache.druid.java.util.common.lifecycle.LifecycleStop;
@@ -42,6 +40,7 @@ import
org.apache.druid.java.util.emitter.service.ServiceEmitter;
import org.apache.druid.metadata.segment.cache.SegmentMetadataCache;
import org.apache.druid.server.coordinator.CoordinatorOverlordServiceConfig;
+import java.util.Set;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantLock;
@@ -87,8 +86,7 @@ public class DruidOverlord
@IndexingService final DruidLeaderSelector overlordLeaderSelector,
final SegmentAllocationQueue segmentAllocationQueue,
final SegmentMetadataCache segmentMetadataCache,
- final CompactionScheduler compactionScheduler,
- final ScheduledBatchTaskManager scheduledBatchTaskManager,
+ final Set<LeaderOverlordService> overlordServices,
final ObjectMapper mapper,
final TaskContextEnricher taskContextEnricher
)
@@ -161,8 +159,7 @@ public class DruidOverlord
public void start()
{
taskMaster.becomeFullLeader();
- compactionScheduler.becomeLeader();
- scheduledBatchTaskManager.start();
+
overlordServices.forEach(LeaderOverlordService::becomeLeader);
// Mark ready only after all the services have been
initialized
initialized = true;
@@ -171,8 +168,7 @@ public class DruidOverlord
@Override
public void stop()
{
- scheduledBatchTaskManager.stop();
- compactionScheduler.stopBeingLeader();
+
overlordServices.forEach(LeaderOverlordService::stopBeingLeader);
taskMaster.downgradeToHalfLeader();
}
}
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManager.java
b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManager.java
index d99fcfa9e68..507081c8340 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManager.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManager.java
@@ -435,18 +435,15 @@ public class SupervisorManager implements
SupervisorStatsProvider
Preconditions.checkState(started, "SupervisorManager not started");
Preconditions.checkNotNull(id, "id");
- Pair<Supervisor, SupervisorSpec> supervisor = supervisors.get(id);
-
- if (supervisor == null) {
- throw new IAE("Supervisor[%s] does not exist", id);
- }
-
- if (!(supervisor.lhs instanceof SeekableStreamSupervisor)) {
- throw new IAE("Supervisor[%s] is not a streaming supervisor", id);
- }
+ Pair<SeekableStreamSupervisor, SeekableStreamSupervisorSpec> supervisor =
getSupervisorOfType(
+ id,
+ SeekableStreamSupervisor.class,
+ SeekableStreamSupervisorSpec.class,
+ "resetToLatestAndBackfill"
+ );
- SeekableStreamSupervisor streamSupervisor = (SeekableStreamSupervisor)
supervisor.lhs;
- SeekableStreamSupervisorSpec streamSpec = (SeekableStreamSupervisorSpec)
supervisor.rhs;
+ SeekableStreamSupervisor streamSupervisor = supervisor.lhs;
+ SeekableStreamSupervisorSpec streamSpec = supervisor.rhs;
validateResetAndBackfill(id, streamSupervisor, streamSpec);
@@ -747,9 +744,29 @@ public class SupervisorManager implements
SupervisorStatsProvider
private StreamSupervisor requireStreamSupervisor(final String supervisorId,
final String operation)
{
- Pair<Supervisor, SupervisorSpec> supervisor =
supervisors.get(supervisorId);
- if (supervisor.lhs instanceof StreamSupervisor) {
- return (StreamSupervisor) supervisor.lhs;
+ return getSupervisorOfType(supervisorId, StreamSupervisor.class,
SupervisorSpec.class, operation).lhs;
+ }
+
+ /**
+ * Finds the non-null supervisor for the given ID only and its corresponding
+ * spec only if they are of the specified type.
+ *
+ * @throws DruidException if the supervisor does not exist or is not of the
+ * specified type.
+ */
+ @SuppressWarnings("unchecked")
+ public <S extends Supervisor, T extends SupervisorSpec> Pair<S, T>
getSupervisorOfType(
+ String supervisorId,
+ Class<S> supervisorType,
+ Class<T> supervisorSpecType,
+ String operation
+ )
+ {
+ final Pair<Supervisor, SupervisorSpec> supervisor =
supervisors.get(supervisorId);
+ if (supervisor == null) {
+ throw NotFound.exception("Supervisor[%s] does not exist", supervisorId);
+ } else if (supervisorType.isInstance(supervisor.lhs) &&
supervisorSpecType.isInstance(supervisor.rhs)) {
+ return (Pair<S, T>) supervisor;
} else {
throw DruidException.forPersona(DruidException.Persona.USER)
.ofCategory(DruidException.Category.UNSUPPORTED)
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/scheduledbatch/ScheduledBatchTaskManager.java
b/indexing-service/src/main/java/org/apache/druid/indexing/scheduledbatch/ScheduledBatchTaskManager.java
index d87b5130904..860710220e2 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/scheduledbatch/ScheduledBatchTaskManager.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/scheduledbatch/ScheduledBatchTaskManager.java
@@ -25,6 +25,7 @@ import org.apache.druid.client.broker.BrokerClient;
import org.apache.druid.common.guava.FutureUtils;
import org.apache.druid.indexer.TaskLocation;
import org.apache.druid.indexer.TaskStatus;
+import org.apache.druid.indexing.overlord.LeaderOverlordService;
import org.apache.druid.indexing.overlord.TaskMaster;
import org.apache.druid.indexing.overlord.TaskRunner;
import org.apache.druid.indexing.overlord.TaskRunnerListener;
@@ -63,7 +64,7 @@ import java.util.concurrent.TimeUnit;
* and is not persisted in the metadata store.
* </p>
*/
-public class ScheduledBatchTaskManager
+public class ScheduledBatchTaskManager implements LeaderOverlordService
{
private static final Logger log = new
EmittingLogger(ScheduledBatchTaskManager.class);
@@ -155,11 +156,9 @@ public class ScheduledBatchTaskManager
/**
* Starts the scheduled batch task manager by registering the {@link
TaskRunnerListener}.
* This allows tracking of any tasks submitted by the batch supervisor.
- * <p>
- * Should be invoked when the Overlord service starts or during leadership
transitions.
- * </p>
*/
- public void start()
+ @Override
+ public void becomeLeader()
{
log.info("Starting scheduled batch task manager.");
final Optional<TaskRunner> taskRunnerOptional = taskMaster.getTaskRunner();
@@ -173,11 +172,9 @@ public class ScheduledBatchTaskManager
/**
* Stops the scheduled batch task manager by shutting down all scheduled
batch supervisors and
* unregistering the registered {@link TaskRunnerListener}.
- * <p>
- * Should be invoked when the Overlord service stops or during leadership
transitions.
- * </p>
*/
- public void stop()
+ @Override
+ public void stopBeingLeader()
{
log.info("Stopping scheduled batch task manager.");
supervisorToTaskScheduler.forEach((supervisorId, taskScheduler) -> {
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/http/OverlordTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/http/OverlordTest.java
index 1c389351709..e7dbc561bee 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/http/OverlordTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/http/OverlordTest.java
@@ -97,6 +97,7 @@ import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
@@ -256,8 +257,10 @@ public class OverlordTest
new TestDruidLeaderSelector(),
EasyMock.createNiceMock(SegmentAllocationQueue.class),
EasyMock.createNiceMock(SegmentMetadataCache.class),
- EasyMock.createNiceMock(CompactionScheduler.class),
- EasyMock.createNiceMock(ScheduledBatchTaskManager.class),
+ Set.of(
+ EasyMock.createNiceMock(CompactionScheduler.class),
+ EasyMock.createNiceMock(ScheduledBatchTaskManager.class)
+ ),
new DefaultObjectMapper(),
new NoopTaskContextEnricher()
);
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManagerTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManagerTest.java
index 7fbb130d2b0..7ef1c85f565 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManagerTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManagerTest.java
@@ -1264,11 +1264,14 @@ public class SupervisorManagerTest extends
EasyMockSupport
{
}
};
+ EasyMock.expect(streamSpec.getType()).andReturn("stream").anyTimes();
+ EasyMock.replay(streamSpec);
supervisorsMap.put("id1", Pair.of(nonStreamSupervisor, streamSpec));
Assert.assertThrows(
- IllegalArgumentException.class,
+ DruidException.class,
() -> manager.resetToLatestAndBackfill("id1", null)
);
+ EasyMock.reset(streamSpec);
// useEarliestSequenceNumber=true → IAE
supervisorsMap.put("id1", Pair.of(streamSupervisor, streamSpec));
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/scheduledbatch/ScheduledBatchTaskManagerTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/scheduledbatch/ScheduledBatchTaskManagerTest.java
index 933703aa995..64b97b44990 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/scheduledbatch/ScheduledBatchTaskManagerTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/scheduledbatch/ScheduledBatchTaskManagerTest.java
@@ -126,7 +126,7 @@ public class ScheduledBatchTaskManagerTest
Mockito.when(brokerClient.submitSqlTask(query1))
.thenReturn(Futures.immediateFuture(expectedTaskStatus));
- scheduler.start();
+ scheduler.becomeLeader();
scheduler.startScheduledIngestion(SUPERVISOR_ID_FOO, DATASOURCE,
IMMEDIATE_SCHEDULER_CONFIG, query1);
verifySchedulerState(SUPERVISOR_ID_FOO,
ScheduledBatchSupervisor.State.RUNNING);
@@ -144,7 +144,7 @@ public class ScheduledBatchTaskManagerTest
ImmutableList.of(TaskStatus.success(expectedTaskStatus.getTaskId()))
);
- scheduler.stop();
+ scheduler.stopBeingLeader();
assertNull(scheduler.getSupervisorStatus(SUPERVISOR_ID_FOO));
serviceEmitter.verifyEmitted(
"task/scheduledBatch/submit/success",
@@ -167,7 +167,7 @@ public class ScheduledBatchTaskManagerTest
)
));
- scheduler.start();
+ scheduler.becomeLeader();
scheduler.startScheduledIngestion(SUPERVISOR_ID_FOO, DATASOURCE,
IMMEDIATE_SCHEDULER_CONFIG, query1);
verifySchedulerState(SUPERVISOR_ID_FOO,
ScheduledBatchSupervisor.State.RUNNING);
@@ -178,7 +178,7 @@ public class ScheduledBatchTaskManagerTest
ScheduledBatchSupervisor.State.SUSPENDED
);
- scheduler.stop();
+ scheduler.stopBeingLeader();
assertNull(scheduler.getSupervisorStatus(SUPERVISOR_ID_FOO));
serviceEmitter.verifyEmitted(
"task/scheduledBatch/submit/failed",
@@ -194,7 +194,7 @@ public class ScheduledBatchTaskManagerTest
Mockito.when(brokerClient.submitSqlTask(query1))
.thenReturn(Futures.immediateFuture(expectedTaskStatus));
- scheduler.start();
+ scheduler.becomeLeader();
scheduler.startScheduledIngestion(SUPERVISOR_ID_FOO, DATASOURCE,
IMMEDIATE_SCHEDULER_CONFIG, query1);
verifySchedulerState(SUPERVISOR_ID_FOO,
ScheduledBatchSupervisor.State.RUNNING);
@@ -222,7 +222,7 @@ public class ScheduledBatchTaskManagerTest
);
executor.finishNextPendingTasks(1);
- scheduler.stop();
+ scheduler.stopBeingLeader();
assertNull(scheduler.getSupervisorStatus(SUPERVISOR_ID_FOO));
serviceEmitter.verifyEmitted(
"task/scheduledBatch/submit/success",
@@ -238,7 +238,7 @@ public class ScheduledBatchTaskManagerTest
Mockito.when(brokerClient.submitSqlTask(query1))
.thenReturn(Futures.immediateFuture(expectedTaskStatus));
- scheduler.start();
+ scheduler.becomeLeader();
scheduler.startScheduledIngestion(SUPERVISOR_ID_FOO, DATASOURCE,
IMMEDIATE_SCHEDULER_CONFIG, query1);
verifySchedulerState(SUPERVISOR_ID_FOO,
ScheduledBatchSupervisor.State.RUNNING);
@@ -258,7 +258,7 @@ public class ScheduledBatchTaskManagerTest
);
assertFalse(executor.hasPendingTasks());
- scheduler.stop();
+ scheduler.stopBeingLeader();
assertNull(scheduler.getSupervisorStatus(SUPERVISOR_ID_FOO));
serviceEmitter.verifyEmitted(
"task/scheduledBatch/submit/success",
@@ -285,7 +285,7 @@ public class ScheduledBatchTaskManagerTest
assertFalse(executor.hasPendingTasks());
- scheduler.start();
+ scheduler.becomeLeader();
scheduler.startScheduledIngestion(SUPERVISOR_ID_FOO, DATASOURCE,
IMMEDIATE_SCHEDULER_CONFIG, query1);
scheduler.startScheduledIngestion(SUPERVISOR_ID_BAR, DATASOURCE,
IMMEDIATE_SCHEDULER_CONFIG, query2);
@@ -314,7 +314,7 @@ public class ScheduledBatchTaskManagerTest
ImmutableList.of(TaskStatus.failure(TASK_ID_BAR1, null),
TaskStatus.success(TASK_ID_BAR2))
);
- scheduler.stop();
+ scheduler.stopBeingLeader();
assertNull(scheduler.getSupervisorStatus(SUPERVISOR_ID_FOO));
assertNull(scheduler.getSupervisorStatus(SUPERVISOR_ID_BAR));
serviceEmitter.verifyEmitted(
diff --git
a/multi-stage-query/src/main/java/org/apache/druid/msq/dart/guice/DartWorkerModule.java
b/multi-stage-query/src/main/java/org/apache/druid/msq/dart/guice/DartWorkerModule.java
index 5ee9aa1f6ee..406c2acac84 100644
---
a/multi-stage-query/src/main/java/org/apache/druid/msq/dart/guice/DartWorkerModule.java
+++
b/multi-stage-query/src/main/java/org/apache/druid/msq/dart/guice/DartWorkerModule.java
@@ -78,9 +78,14 @@ import java.util.concurrent.ExecutorService;
@LoadScope(roles = NodeRole.HISTORICAL_JSON_NAME)
public class DartWorkerModule implements DruidModule
{
- @Inject
private Properties properties;
+ @Inject
+ public void setProperties(Properties properties)
+ {
+ this.properties = properties;
+ }
+
@Override
public void configure(Binder binder)
{
diff --git
a/server/src/main/java/org/apache/druid/indexing/overlord/LeaderOverlordService.java
b/server/src/main/java/org/apache/druid/indexing/overlord/LeaderOverlordService.java
new file mode 100644
index 00000000000..738a0c91ee0
--- /dev/null
+++
b/server/src/main/java/org/apache/druid/indexing/overlord/LeaderOverlordService.java
@@ -0,0 +1,58 @@
+/*
+ * 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.indexing.overlord;
+
+/**
+ * Represents a stateful service running on the leader Overlord. A single
service
+ * is responsible for a particular aspect of Overlord operations such as
+ * managing segment allocation, launching compaction jobs, etc.
+ * <p>
+ * This is distinct from an {@code OverlordDuty} which contains relatively
simpler
+ * operations that run on the {@code OverlordDutyExecutor} itself. On the other
+ * hand, a {@link LeaderOverlordService} may have its own dedicated thread
pool.
+ */
+public interface LeaderOverlordService
+{
+ /**
+ * Called when this Overlord becomes leader so that the service can
initialize
+ * state and start its scheduled management.
+ * <p>
+ * The order in which this method is called for different instances of
+ * {@link LeaderOverlordService} is non-deterministic. However, it is always
+ * called after initializing all other Overlord dependencies.
+ * <p>
+ * This method blocks the Overlord lifecycle thread and thus must not be used
+ * to perform long computations.
+ */
+ void becomeLeader();
+
+ /**
+ * Called when this Overlord is not leader anymore so that the service can
+ * interrupt any ongoing tasks and clean up state.
+ * <p>
+ * The order in which this method is called for different instances of
+ * {@link LeaderOverlordService} is non-deterministic. However, it is always
+ * called after initializing all other Overlord dependencies.
+ * <p>
+ * This method blocks the Overlord lifecycle thread and thus must not be used
+ * to perform long computations.
+ */
+ void stopBeingLeader();
+}
diff --git a/services/src/main/java/org/apache/druid/cli/CliOverlord.java
b/services/src/main/java/org/apache/druid/cli/CliOverlord.java
index f7e2d95d9b6..9b81a4a987b 100644
--- a/services/src/main/java/org/apache/druid/cli/CliOverlord.java
+++ b/services/src/main/java/org/apache/druid/cli/CliOverlord.java
@@ -77,6 +77,7 @@ import
org.apache.druid.indexing.overlord.ForkingTaskRunnerFactory;
import org.apache.druid.indexing.overlord.GlobalTaskLockbox;
import org.apache.druid.indexing.overlord.HeapMemoryTaskStorage;
import org.apache.druid.indexing.overlord.IndexerMetadataStorageAdapter;
+import org.apache.druid.indexing.overlord.LeaderOverlordService;
import org.apache.druid.indexing.overlord.MetadataTaskStorage;
import org.apache.druid.indexing.overlord.TaskMaster;
import org.apache.druid.indexing.overlord.TaskQueryTool;
@@ -261,6 +262,12 @@ public class CliOverlord extends ServerRunnable
binder.bind(ShuffleClient.class).toProvider(Providers.of(null));
binder.bind(ChatHandlerProvider.class).in(LazySingleton.class);
+ // Bind the schedulers as impls of LeaderOverlordService
+ final Multibinder<LeaderOverlordService> leaderServiceBinder =
+ Multibinder.newSetBinder(binder, LeaderOverlordService.class);
+ leaderServiceBinder.addBinding().to(CompactionScheduler.class);
+
leaderServiceBinder.addBinding().to(ScheduledBatchTaskManager.class);
+
CliPeon.bindDataSegmentKiller(binder);
PolyBind.createChoice(
diff --git
a/services/src/test/java/org/apache/druid/testing/embedded/EmbeddedServiceClient.java
b/services/src/test/java/org/apache/druid/testing/embedded/EmbeddedServiceClient.java
index 2621751b0b7..b324d2236a2 100644
---
a/services/src/test/java/org/apache/druid/testing/embedded/EmbeddedServiceClient.java
+++
b/services/src/test/java/org/apache/druid/testing/embedded/EmbeddedServiceClient.java
@@ -51,7 +51,6 @@ import org.apache.druid.rpc.guice.ServiceClientModule;
import org.apache.druid.rpc.indexing.OverlordClient;
import org.apache.druid.server.security.Escalator;
import org.apache.druid.sql.http.ResultFormat;
-import org.jboss.netty.handler.codec.http.HttpResponseStatus;
import javax.annotation.Nullable;
import java.util.concurrent.ScheduledExecutorService;
@@ -266,8 +265,10 @@ public class EmbeddedServiceClient
try {
StatusResponseHolder response = serviceClient.request(requestBuilder,
responseHandler);
- if (!response.getStatus().equals(HttpResponseStatus.OK)
- && !response.getStatus().equals(HttpResponseStatus.ACCEPTED)) {
+
+ // Handle all success status codes
+ final int statusCode = response.getStatus().getCode();
+ if (statusCode < 200 || statusCode >= 300) {
throw new ISE(
"Request[%s] failed with status[%s] content[%s].",
requestBuilder.toString(),
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
