This is an automated email from the ASF dual-hosted git repository.
abhishekrb19 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 cf75e06d4f9 fix: Prevent a single unhealthy historical from stalling
all segment loading and balancing. (#19950)
cf75e06d4f9 is described below
commit cf75e06d4f9343fe750235f69abf7b15a81c97c8
Author: Abhishek Radhakrishnan <[email protected]>
AuthorDate: Tue Aug 25 09:34:45 2026 -0700
fix: Prevent a single unhealthy historical from stalling all segment
loading and balancing. (#19950)
HttpLoadQueuePeon fetches a historical's segment-loading capabilities
synchronously in its constructor, via GET
/druid-internal/v1/segments/loadCapabilities.
On any response other than 200 or 404 (e.g. a 503 from an overloaded or
briefly-unavailable historical), the constructor threw an RE.
Make the capabilities fetch degrade gracefully instead of throwing. On a
non-OK
response or any error, alert and fall back to default
SegmentLoadingCapabilities
(derived from the configured batch size) — mirroring the behavior the 404
branch
already had. The peon is still created, the server is still managed with
conservative defaults, and the rest of the duty group proceeds. Operators
still
get one alert per affected server.
---
.../coordinator/loading/HttpLoadQueuePeon.java | 196 +++++++++++++---
.../server/coordinator/DruidCoordinatorTest.java | 210 +++++++++++++++++
.../coordinator/loading/HttpLoadQueuePeonTest.java | 261 +++++++++++++++++++++
3 files changed, 639 insertions(+), 28 deletions(-)
diff --git
a/server/src/main/java/org/apache/druid/server/coordinator/loading/HttpLoadQueuePeon.java
b/server/src/main/java/org/apache/druid/server/coordinator/loading/HttpLoadQueuePeon.java
index 37ebaa8d9a0..438bea3a7bd 100644
---
a/server/src/main/java/org/apache/druid/server/coordinator/loading/HttpLoadQueuePeon.java
+++
b/server/src/main/java/org/apache/druid/server/coordinator/loading/HttpLoadQueuePeon.java
@@ -57,7 +57,9 @@ import org.joda.time.Duration;
import javax.annotation.Nullable;
import javax.servlet.http.HttpServletResponse;
import javax.ws.rs.core.MediaType;
+import java.io.IOException;
import java.io.InputStream;
+import java.net.MalformedURLException;
import java.net.URL;
import java.util.ArrayList;
import java.util.Collections;
@@ -127,7 +129,32 @@ public class HttpLoadQueuePeon implements LoadQueuePeon
private final Supplier<SegmentLoadingMode> loadingModeSupplier;
private final ObjectWriter requestBodyWriter;
- private final SegmentLoadingCapabilities serverCapabilities;
+
+ /**
+ * Loading capabilities of the server. Fetched during construction and
re-fetched
+ * lazily on subsequent ticks if the initial fetch fell back to default
values due
+ * to a transient failure (see {@link #refetchCapabilitiesIfNeeded()}). Read
and
+ * written only on the single-threaded processing executor after
construction, but
+ * declared {@code volatile} for safe publication from the constructing
thread.
+ */
+ private volatile SegmentLoadingCapabilities serverCapabilities;
+
+ /**
+ * Whether {@link #serverCapabilities} holds a definitive value: {@code
true} once
+ * the server has returned real capabilities, or a 404 indicating the
endpoint does
+ * not exist on this server. It stays {@code false} while the peon is pinned
to
+ * default capabilities due to a transient failure, so the value is
re-fetched once
+ * the server recovers.
+ */
+ private volatile boolean capabilitiesConfirmed = false;
+
+ /**
+ * Guards {@link #refetchCapabilitiesIfNeeded()} so that at most one
capability probe is
+ * outstanding at a time. Without this, every {@link #doSegmentManagement()}
tick issues its
+ * own probe whenever {@link #capabilitiesConfirmed} is false, so queuing
many segments onto a
+ * still-unhealthy server fires one redundant concurrent probe per tick at
that same server.
+ */
+ private final AtomicBoolean refetchInProgress = new AtomicBoolean(false);
public HttpLoadQueuePeon(
String baseUrl,
@@ -151,48 +178,160 @@ public class HttpLoadQueuePeon implements LoadQueuePeon
this.serverCapabilities = fetchSegmentLoadingCapabilities();
}
+ private URL getLoadCapabilitiesUrl() throws MalformedURLException
+ {
+ return new URL(new URL(serverId),
"druid-internal/v1/segments/loadCapabilities");
+ }
+
+ /**
+ * Synchronously fetches loading capabilities during construction. On a
transient failure
+ * (non-OK status other than 404, timeout, or error), raises an alert and
falls back to
+ * default capabilities, leaving {@link #capabilitiesConfirmed} unset so the
value is
+ * re-fetched on a later tick once the server recovers (see {@link
#refetchCapabilitiesIfNeeded()}).
+ */
private SegmentLoadingCapabilities fetchSegmentLoadingCapabilities()
{
try {
- final URL segmentLoadingCapabilitiesURL = new URL(
- new URL(serverId),
- "druid-internal/v1/segments/loadCapabilities"
- );
-
- BytesAccumulatingResponseHandler responseHandler = new
BytesAccumulatingResponseHandler();
- InputStream stream = httpClient.go(
- new Request(HttpMethod.GET, segmentLoadingCapabilitiesURL)
- .addHeader(HttpHeaders.Names.ACCEPT, MediaType.APPLICATION_JSON),
+ final URL url = getLoadCapabilitiesUrl();
+ final BytesAccumulatingResponseHandler responseHandler = new
BytesAccumulatingResponseHandler();
+ final InputStream stream = httpClient.go(
+ new Request(HttpMethod.GET, url).addHeader(HttpHeaders.Names.ACCEPT,
MediaType.APPLICATION_JSON),
responseHandler,
new Duration(DEFAULT_TIMEOUT)
).get();
- if (HttpServletResponse.SC_NOT_FOUND == responseHandler.getStatus()) {
- int batchSize = config.getBatchSize() == null ? 1 :
config.getBatchSize();
- SegmentLoadingCapabilities defaultCapabilities = new
SegmentLoadingCapabilities(batchSize, batchSize);
- log.warn(
- "Historical capabilities endpoint not found at URL[%s]. Using
default values[%s].",
- segmentLoadingCapabilitiesURL,
- defaultCapabilities
- );
- return defaultCapabilities;
- } else if (HttpServletResponse.SC_OK != responseHandler.getStatus()) {
- log.makeAlert("Received status[%s] when fetching loading capabilities
from server[%s]", responseHandler.getStatus(), serverId);
- throw new RE("Received status[%s] when fetching loading capabilities
from server[%s]", responseHandler.getStatus(), serverId);
+ final int status = responseHandler.getStatus();
+ final SegmentLoadingCapabilities capabilities =
interpretCapabilitiesResponse(status, stream, url);
+ if (!capabilitiesConfirmed) {
+ // Transient failure. Do not stall further processing due to a single
unhealthy server:
+ // raise an alert and use default capabilities until the server
recovers.
+ log.makeAlert(
+ "Received status[%s] when fetching loading capabilities from
server[%s]. Using default values[%s].",
+ status,
+ serverId,
+ capabilities
+ ).emit();
}
+ return capabilities;
+ }
+ catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(ie);
+ }
+ catch (Exception e) {
+ SegmentLoadingCapabilities defaultCapabilities =
getDefaultLoadingCapabilities();
+ log.makeAlert(
+ e,
+ "Received error while fetching historical capabilities from
Server[%s]. Using default values[%s].",
+ serverId,
+ defaultCapabilities
+ ).emit();
+ return defaultCapabilities;
+ }
+ }
+
+ /**
+ * Interprets a loadCapabilities response, setting {@link
#capabilitiesConfirmed} and returning
+ * the capabilities to use. The value is confirmed on a real response (200)
or a 404 (the endpoint
+ * is absent on this server, so retrying is pointless). A transient non-OK
status yields default
+ * capabilities without confirming, so they are re-fetched on a later tick
once the server recovers.
+ */
+ private SegmentLoadingCapabilities interpretCapabilitiesResponse(int status,
InputStream stream, URL url)
+ throws IOException
+ {
+ if (HttpServletResponse.SC_NOT_FOUND == status) {
+ capabilitiesConfirmed = true;
+ SegmentLoadingCapabilities defaultCapabilities =
getDefaultLoadingCapabilities();
+ log.warn(
+ "Historical capabilities endpoint not found at URL[%s]. Using
default values[%s].",
+ url,
+ defaultCapabilities
+ );
+ return defaultCapabilities;
+ } else if (HttpServletResponse.SC_OK != status) {
+ return getDefaultLoadingCapabilities();
+ }
+
+ SegmentLoadingCapabilities capabilities = jsonMapper.readValue(stream,
SegmentLoadingCapabilities.class);
+ capabilitiesConfirmed = true;
+ return capabilities;
+ }
+
+ /**
+ * Re-fetches loading capabilities if the peon is still pinned to default
values from a
+ * transient failure during construction. Called on every segment management
tick; a no-op
+ * once capabilities have been confirmed (the common case).
+ * <p>
+ * Unlike the construction path, this does not block the (single, shared)
processing thread:
+ * the request is issued and its response handled in a callback, just like
the segment change
+ * requests in {@link #doSegmentManagement()}. This keeps a single unhealthy
server from
+ * stalling segment management for the rest of the cluster.
+ */
+ private void refetchCapabilitiesIfNeeded()
+ {
+ if (capabilitiesConfirmed || stopped ||
!refetchInProgress.compareAndSet(false, true)) {
+ return;
+ }
+
+ try {
+ final URL url = getLoadCapabilitiesUrl();
+ final BytesAccumulatingResponseHandler responseHandler = new
BytesAccumulatingResponseHandler();
+ final ListenableFuture<InputStream> future = httpClient.go(
+ new Request(HttpMethod.GET, url).addHeader(HttpHeaders.Names.ACCEPT,
MediaType.APPLICATION_JSON),
+ responseHandler,
+ new Duration(DEFAULT_TIMEOUT)
+ );
+
+ Futures.addCallback(
+ future,
+ new FutureCallback<>()
+ {
+ @Override
+ public void onSuccess(InputStream result)
+ {
+ try {
+ serverCapabilities =
interpretCapabilitiesResponse(responseHandler.getStatus(), result, url);
+ if (capabilitiesConfirmed) {
+ log.info("Refreshed loading capabilities[%s] for
server[%s].", serverCapabilities, serverId);
+ }
+ }
+ catch (Throwable t) {
+ log.debug(t, "Could not parse refreshed loading capabilities
from server[%s]. Will retry.", serverId);
+ }
+ finally {
+ refetchInProgress.set(false);
+ }
+ }
- return jsonMapper.readValue(
- stream,
- SegmentLoadingCapabilities.class
+ @Override
+ public void onFailure(Throwable t)
+ {
+ log.debug(t, "Could not refresh loading capabilities from
server[%s]. Will retry.", serverId);
+ refetchInProgress.set(false);
+ }
+ },
+ processingExecutor
);
}
catch (Throwable th) {
- throw new RE(th, "Received error while fetching historical capabilities
from Server[%s].", serverId);
+ log.debug(th, "Error issuing capability refresh request to server[%s].
Will retry.", serverId);
+ refetchInProgress.set(false);
}
}
+ private SegmentLoadingCapabilities getDefaultLoadingCapabilities()
+ {
+ int batchSize = config.getBatchSize() == null ? 1 : config.getBatchSize();
+ return new SegmentLoadingCapabilities(batchSize, batchSize);
+ }
+
private void doSegmentManagement()
{
+ // Re-fetch loading capabilities if we are still pinned to defaults from a
transient
+ // failure. This is async and a no-op once capabilities are confirmed, so
it runs
+ // independently of the main loop below (which may bail out early if
already in progress).
+ refetchCapabilitiesIfNeeded();
+
if (stopped || !mainLoopInProgress.compareAndSet(false, true)) {
log.trace("[%s]Ignoring tick. Either in-progress already or stopped.",
serverId);
return;
@@ -371,11 +510,12 @@ public class HttpLoadQueuePeon implements LoadQueuePeon
@VisibleForTesting
int calculateBatchSize(SegmentLoadingMode loadingMode)
{
+ final SegmentLoadingCapabilities capabilities = serverCapabilities;
int batchSize;
if (SegmentLoadingMode.TURBO.equals(loadingMode)) {
- batchSize = serverCapabilities.getNumTurboLoadingThreads();
+ batchSize = capabilities.getNumTurboLoadingThreads();
} else {
- batchSize = Configs.valueOrDefault(config.getBatchSize(),
serverCapabilities.getNumLoadingThreads());
+ batchSize = Configs.valueOrDefault(config.getBatchSize(),
capabilities.getNumLoadingThreads());
}
return Math.max(batchSize, 1);
diff --git
a/server/src/test/java/org/apache/druid/server/coordinator/DruidCoordinatorTest.java
b/server/src/test/java/org/apache/druid/server/coordinator/DruidCoordinatorTest.java
index efa67668e8d..6e2c2a8f73f 100644
---
a/server/src/test/java/org/apache/druid/server/coordinator/DruidCoordinatorTest.java
+++
b/server/src/test/java/org/apache/druid/server/coordinator/DruidCoordinatorTest.java
@@ -19,9 +19,12 @@
package org.apache.druid.server.coordinator;
+import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
import it.unimi.dsi.fastutil.objects.Object2IntMap;
import it.unimi.dsi.fastutil.objects.Object2LongMap;
import org.apache.druid.client.DataSourcesSnapshot;
@@ -33,16 +36,23 @@ import org.apache.druid.client.ServerInventoryView;
import org.apache.druid.common.config.JacksonConfigManager;
import org.apache.druid.discovery.DruidLeaderSelector;
import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.java.util.common.concurrent.Execs;
import org.apache.druid.java.util.common.concurrent.ScheduledExecutorFactory;
import org.apache.druid.java.util.common.concurrent.ScheduledExecutors;
+import org.apache.druid.java.util.emitter.EmittingLogger;
import org.apache.druid.java.util.emitter.core.Event;
+import org.apache.druid.java.util.emitter.service.AlertEvent;
import org.apache.druid.java.util.emitter.service.ServiceEmitter;
import org.apache.druid.java.util.emitter.service.ServiceMetricEvent;
+import org.apache.druid.java.util.http.client.HttpClient;
+import org.apache.druid.java.util.http.client.Request;
+import org.apache.druid.java.util.http.client.response.HttpResponseHandler;
import org.apache.druid.metadata.MetadataRuleManager;
import org.apache.druid.metadata.MetadataRuleManagerConfig;
import org.apache.druid.metadata.SegmentsMetadataManager;
import org.apache.druid.metadata.segment.cache.NoopSegmentMetadataCache;
import org.apache.druid.rpc.indexing.OverlordClient;
+import org.apache.druid.segment.TestHelper;
import org.apache.druid.segment.metadata.CentralizedDatasourceSchemaConfig;
import org.apache.druid.server.DruidNode;
import org.apache.druid.server.compaction.CompactionSimulateResult;
@@ -53,6 +63,7 @@ import
org.apache.druid.server.coordinator.config.CoordinatorKillConfigs;
import org.apache.druid.server.coordinator.config.CoordinatorPeriodConfig;
import org.apache.druid.server.coordinator.config.CoordinatorRunConfig;
import org.apache.druid.server.coordinator.config.DruidCoordinatorConfig;
+import org.apache.druid.server.coordinator.config.HttpLoadQueuePeonConfig;
import org.apache.druid.server.coordinator.config.MetadataCleanupConfig;
import org.apache.druid.server.coordinator.duty.CompactSegments;
import org.apache.druid.server.coordinator.duty.CoordinatorCustomDuty;
@@ -74,9 +85,15 @@ import org.apache.druid.server.coordinator.rules.Rule;
import org.apache.druid.server.coordinator.stats.Stats;
import org.apache.druid.server.http.BrokerDynamicConfigSyncer;
import org.apache.druid.server.http.CoordinatorDynamicConfigSyncer;
+import org.apache.druid.server.http.SegmentLoadingCapabilities;
import org.apache.druid.server.lookup.cache.LookupCoordinatorManager;
import org.apache.druid.timeline.DataSegment;
import org.easymock.EasyMock;
+import org.jboss.netty.buffer.ChannelBuffers;
+import org.jboss.netty.handler.codec.http.DefaultHttpResponse;
+import org.jboss.netty.handler.codec.http.HttpResponse;
+import org.jboss.netty.handler.codec.http.HttpResponseStatus;
+import org.jboss.netty.handler.codec.http.HttpVersion;
import org.joda.time.Duration;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
@@ -85,11 +102,14 @@ import org.junit.jupiter.api.Timeout;
import org.junit.jupiter.api.Timeout.ThreadMode;
import javax.annotation.Nullable;
+import java.io.ByteArrayInputStream;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
@@ -908,6 +928,173 @@ public class DruidCoordinatorTest
};
}
+ /**
+ * End-to-end check via a real {@link DruidCoordinator} and {@link
LoadQueueTaskMaster}:
+ * a single server returning HTTP 503 on {@code loadCapabilities} must not
abort the
+ * {@code HistoricalManagementDuties} group. Asserts the group completes a
run, no
+ * capabilities-related failure alert is emitted, and the healthy servers
are still managed.
+ */
+ @Test
+ @Timeout(value = 60_000L, unit = TimeUnit.MILLISECONDS, threadMode =
ThreadMode.SEPARATE_THREAD)
+ public void testUnhealthyHistoricalDoesNotAbortDutyGroup() throws Exception
+ {
+ EmittingLogger.registerEmitter(serviceEmitter);
+
+ EasyMock.expect(metadataRuleManager.getRulesSnapshot())
+ .andReturn(clusterDefaultRules(new
ForeverLoadRule(ImmutableMap.of("tier1", 1), null)))
+ .anyTimes();
+ EasyMock.replay(metadataRuleManager);
+
+ final DruidDataSource dataSource = new DruidDataSource("ds",
Collections.emptyMap());
+ dataSource.addSegment(new DataSegment("ds",
Intervals.of("2010-01-01/P1D"), "v1", null, null, null, null, 0x9, 0));
+ setupSegmentsMetadataMock(dataSource);
+
+ // A cluster of healthy historicals plus a single unhealthy one that 503s
on the
+ // loadCapabilities endpoint. The healthy servers must keep being managed.
+ final DruidServer healthy1 = new DruidServer("healthy1", "healthy1:8088",
null, 100L, null, ServerType.HISTORICAL, "tier1", 0);
+ final DruidServer unhealthy = new DruidServer("unhealthy",
"unhealthy:8088", null, 100L, null, ServerType.HISTORICAL, "tier1", 0);
+ final DruidServer healthy2 = new DruidServer("healthy2", "healthy2:8088",
null, 100L, null, ServerType.HISTORICAL, "tier1", 0);
+ EasyMock.expect(serverInventoryView.getInventory())
+ .andReturn(ImmutableList.of(healthy1, unhealthy, healthy2))
+ .anyTimes();
+
EasyMock.expect(serverInventoryView.isStarted()).andReturn(true).anyTimes();
+ EasyMock.replay(serverInventoryView);
+
+ // Build a coordinator wired with a REAL LoadQueueTaskMaster so that peon
+ // creation (and the failing capabilities fetch) actually happens. Only the
+ // unhealthy server 503s; the healthy ones answer normally.
+ final ScheduledExecutorService peonExec =
Execs.scheduledSingleThreaded("Coordinator-peon-%s");
+ final ExecutorService callbackExec =
Execs.singleThreaded("Coordinator-cb-%s");
+ final LoadQueueTaskMaster realTaskMaster = new LoadQueueTaskMaster(
+ TestHelper.makeJsonMapper(),
+ peonExec,
+ callbackExec,
+ new HttpLoadQueuePeonConfig(null, null, 10),
+ new SelectivelyFailingHttpClient("unhealthy"),
+ () -> CoordinatorDynamicConfig.builder().build()
+ );
+
+ final JacksonConfigManager configManager =
EasyMock.createNiceMock(JacksonConfigManager.class);
+
EasyMock.expect(configManager.watch(EasyMock.eq(CoordinatorDynamicConfig.CONFIG_KEY),
EasyMock.anyObject(Class.class), EasyMock.anyObject()))
+ .andReturn(new
AtomicReference<>(CoordinatorDynamicConfig.builder().build())).anyTimes();
+
EasyMock.expect(configManager.watch(EasyMock.eq(DruidCompactionConfig.CONFIG_KEY),
EasyMock.anyObject(Class.class), EasyMock.anyObject()))
+ .andReturn(new
AtomicReference<>(DruidCompactionConfig.empty())).anyTimes();
+ EasyMock.replay(configManager);
+
+ final DruidCoordinator coordinatorWithRealTaskMaster = new
DruidCoordinator(
+ druidCoordinatorConfig,
+ createMetadataManager(configManager),
+ serverInventoryView,
+ serviceEmitter,
+ scheduledExecutorFactory,
+ overlordClient,
+ realTaskMaster,
+ new SegmentLoadQueueManager(serverInventoryView, realTaskMaster),
+ new CoordinatorCustomDutyGroups(ImmutableSet.of()),
+ EasyMock.createNiceMock(LookupCoordinatorManager.class),
+ new TestDruidLeaderSelector(),
+ null,
+ CentralizedDatasourceSchemaConfig.create(),
+ new CompactionStatusTracker(),
+ EasyMock.niceMock(CoordinatorDynamicConfigSyncer.class),
+ EasyMock.niceMock(BrokerDynamicConfigSyncer.class),
+ new CloneStatusManager()
+ );
+
+ try {
+ coordinatorWithRealTaskMaster.start();
+
+ // The HistoricalManagementDuties group must actually complete a run
despite the
+ // sick historical. GROUP_RUN_TIME is emitted only at the end of a full
group run,
+ // so this latch (2 runs) only trips if the group is not being aborted.
+ Assertions.assertTrue(
+ serviceEmitter.coordinatorRunLatch.await(30,
java.util.concurrent.TimeUnit.SECONDS),
+ "HistoricalManagementDuties group did not complete a run; one
unhealthy historical aborted it"
+ );
+
+ // And the coordinator must not have hit the catch-all that logs
+ // "Caught exception, ignoring so that schedule keeps going." for a
failure in the
+ // load-queue/peon-preparation path.
+ Assertions.assertNull(
+ serviceEmitter.dutyGroupFailureAlert.get(),
+ "Coordinator aborted the duty group with an exception for a single
unhealthy historical"
+ );
+
+ Assertions.assertTrue(coordinatorWithRealTaskMaster.isLeader());
+
+ // The healthy servers must still be managed (they have peons); only the
+ // unhealthy one may be missing.
+ Assertions.assertNotNull(
+ realTaskMaster.getPeonForServer(healthy1.toImmutableDruidServer()),
+ "Healthy server healthy1 was never managed because an unhealthy
server aborted the reconcile"
+ );
+ Assertions.assertNotNull(
+ realTaskMaster.getPeonForServer(healthy2.toImmutableDruidServer()),
+ "Healthy server healthy2 was never managed because an unhealthy
server aborted the reconcile"
+ );
+ }
+ finally {
+ coordinatorWithRealTaskMaster.stop();
+ peonExec.shutdownNow();
+ callbackExec.shutdownNow();
+ }
+ }
+
+ /**
+ * HttpClient that answers the {@code loadCapabilities} call, returning HTTP
503 for a
+ * single named server and a valid capabilities payload for everyone else.
+ */
+ private static class SelectivelyFailingHttpClient implements HttpClient
+ {
+ private static final ObjectMapper MAPPER = TestHelper.makeJsonMapper();
+
+ private final String failingServer;
+
+ SelectivelyFailingHttpClient(String failingServer)
+ {
+ this.failingServer = failingServer;
+ }
+
+ @Override
+ public <Intermediate, Final> ListenableFuture<Final> go(
+ Request request,
+ HttpResponseHandler<Intermediate, Final> handler
+ )
+ {
+ return go(request, handler, null);
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public <Intermediate, Final> ListenableFuture<Final> go(
+ Request request,
+ HttpResponseHandler<Intermediate, Final> handler,
+ Duration duration
+ )
+ {
+ final String url = request.getUrl().toString();
+ final boolean isFailing = failingServer != null &&
url.contains(failingServer);
+
+ final HttpResponseStatus status = isFailing
+ ?
HttpResponseStatus.SERVICE_UNAVAILABLE
+ : HttpResponseStatus.OK;
+ final HttpResponse httpResponse = new
DefaultHttpResponse(HttpVersion.HTTP_1_1, status);
+ httpResponse.setContent(ChannelBuffers.buffer(0));
+ handler.handleResponse(httpResponse, null);
+
+ try {
+ final byte[] body = isFailing
+ ? new byte[0]
+ :
MAPPER.writerFor(SegmentLoadingCapabilities.class)
+ .writeValueAsBytes(new
SegmentLoadingCapabilities(1, 3));
+ return (ListenableFuture<Final>) Futures.immediateFuture(new
ByteArrayInputStream(body));
+ }
+ catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
private static class TestDruidLeaderSelector implements DruidLeaderSelector
{
private volatile Listener listener;
@@ -951,6 +1138,13 @@ public class DruidCoordinatorTest
{
private final CountDownLatch coordinatorRunLatch = new CountDownLatch(2);
+ /**
+ * Set when {@code DruidCoordinator.DutiesRunnable.run()} catches an
exception thrown
+ * while preparing the balancer / load queues (peon creation +
capabilities fetch) and
+ * emits its "Caught exception, ignoring so that schedule keeps going."
alert.
+ */
+ private final AtomicReference<String> dutyGroupFailureAlert = new
AtomicReference<>();
+
private LatchableServiceEmitter()
{
super("", "", null);
@@ -968,6 +1162,22 @@ public class DruidCoordinatorTest
&& "HistoricalManagementDuties".equals(dutyGroupName)) {
coordinatorRunLatch.countDown();
}
+ } else if (event instanceof AlertEvent) {
+ final AlertEvent alert = (AlertEvent) event;
+ // The coordinator's top-level catch logs this exact description for a
failure in ANY duty
+ // group. This test's harness intentionally leaves several
collaborators (audit manager,
+ // supervisor manager, etc.) null, so unrelated duty groups such as
IndexingServiceDuties emit
+ // this alert too. Scope detection to the bug under test: an exception
thrown while preparing
+ // the balancer / load queues (peon creation + capabilities fetch) for
HistoricalManagementDuties.
+ if ("Caught exception, ignoring so that schedule keeps
going.".equals(alert.getDescription())) {
+ final Object stackTrace =
alert.getDataMap().get("exceptionStackTrace");
+ if (stackTrace != null
+ && (stackTrace.toString().contains("HttpLoadQueuePeon")
+ ||
stackTrace.toString().contains("PrepareBalancerAndLoadQueues")
+ ||
stackTrace.toString().contains("resetPeonsForNewServers"))) {
+ dutyGroupFailureAlert.set(alert.getDescription());
+ }
+ }
}
}
}
diff --git
a/server/src/test/java/org/apache/druid/server/coordinator/loading/HttpLoadQueuePeonTest.java
b/server/src/test/java/org/apache/druid/server/coordinator/loading/HttpLoadQueuePeonTest.java
index 1520e9f5b35..7ad3049cbd2 100644
---
a/server/src/test/java/org/apache/druid/server/coordinator/loading/HttpLoadQueuePeonTest.java
+++
b/server/src/test/java/org/apache/druid/server/coordinator/loading/HttpLoadQueuePeonTest.java
@@ -24,6 +24,7 @@ import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import org.apache.druid.java.util.common.RE;
import org.apache.druid.java.util.common.granularity.Granularities;
+import org.apache.druid.java.util.emitter.EmittingLogger;
import org.apache.druid.java.util.http.client.HttpClient;
import org.apache.druid.java.util.http.client.Request;
import org.apache.druid.java.util.http.client.response.HttpResponseHandler;
@@ -39,6 +40,7 @@ import
org.apache.druid.server.coordinator.simulate.BlockingExecutorService;
import
org.apache.druid.server.coordinator.simulate.WrappingScheduledExecutorService;
import org.apache.druid.server.http.SegmentLoadingCapabilities;
import org.apache.druid.server.http.SegmentLoadingMode;
+import org.apache.druid.server.metrics.NoopServiceEmitter;
import org.apache.druid.timeline.DataSegment;
import org.jboss.netty.buffer.ChannelBuffers;
import org.jboss.netty.handler.codec.http.DefaultHttpResponse;
@@ -58,6 +60,7 @@ import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
import java.util.stream.Collectors;
@@ -78,6 +81,7 @@ public class HttpLoadQueuePeonTest
@BeforeEach
public void setUp()
{
+ EmittingLogger.registerEmitter(new NoopServiceEmitter());
segmentLoadingCapabilities = new SegmentLoadingCapabilities(1, 3);
httpClient = new TestHttpClient();
httpLoadQueuePeon = new HttpLoadQueuePeon(
@@ -321,6 +325,123 @@ public class HttpLoadQueuePeonTest
);
}
+ @Test
+ public void testPeonIsCreatedWhenServerErrorsOnLoadCapabilities()
+ {
+ // A server returning a non-OK, non-404 status (e.g. 503) on
loadCapabilities must not
+ // fail peon construction. The peon falls back to default capabilities so
the server is
+ // still managed, rather than throwing and aborting segment management for
the cluster.
+ final HttpClient failingClient = new HttpClient()
+ {
+ @Override
+ public <Intermediate, Final> ListenableFuture<Final> go(
+ Request request,
+ HttpResponseHandler<Intermediate, Final> handler
+ )
+ {
+ return go(request, handler, null);
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public <Intermediate, Final> ListenableFuture<Final> go(
+ Request request,
+ HttpResponseHandler<Intermediate, Final> handler,
+ Duration duration
+ )
+ {
+ respond(handler, HttpResponseStatus.SERVICE_UNAVAILABLE);
+ return (ListenableFuture<Final>) Futures.immediateFuture(new
ByteArrayInputStream(new byte[0]));
+ }
+ };
+
+ final HttpLoadQueuePeon peon = new HttpLoadQueuePeon(
+ "http://dummy:4000";,
+ MAPPER,
+ failingClient,
+ new HttpLoadQueuePeonConfig(null, null, 10),
+ () -> SegmentLoadingMode.NORMAL,
+ new WrappingScheduledExecutorService("HttpLoadQueuePeonTest-%s",
httpClient.processingExecutor, true),
+ httpClient.callbackExecutor
+ );
+
+ // Construction succeeded and the peon fell back to default capabilities
derived from the batch
+ // size: turbo loading threads default to the batch size (10) rather than
a fetched value.
+ Assertions.assertEquals(10,
peon.calculateBatchSize(SegmentLoadingMode.TURBO));
+ }
+
+ @Test
+ public void testCapabilitiesAreReFetchedAfterServerRecovers()
+ {
+ // Regression test for the review finding on #19950. If a server is
unhealthy when the peon is
+ // constructed, the peon falls back to default capabilities. Since
LoadQueueTaskMaster reuses the
+ // same peon while the server stays in inventory, the peon must re-fetch
capabilities on a later
+ // tick once the server recovers, rather than staying pinned to the
defaults forever.
+ final AtomicInteger loadCapabilitiesCalls = new AtomicInteger(0);
+ final RecoveringHttpClient recoveringClient = new
RecoveringHttpClient(loadCapabilitiesCalls);
+
+ final HttpLoadQueuePeon peon = new HttpLoadQueuePeon(
+ "http://dummy:4000";,
+ MAPPER,
+ recoveringClient,
+ new HttpLoadQueuePeonConfig(null, null, 10),
+ () -> SegmentLoadingMode.NORMAL,
+ new WrappingScheduledExecutorService("HttpLoadQueuePeonTest-%s",
recoveringClient.processingExecutor, true),
+ recoveringClient.callbackExecutor
+ );
+
+ // The probe issued during construction failed, so the peon falls back to
default capabilities and
+ // reports the default batch size (10) rather than the real turbo
capability (3).
+ Assertions.assertEquals(1, loadCapabilitiesCalls.get());
+ Assertions.assertEquals(10,
peon.calculateBatchSize(SegmentLoadingMode.TURBO));
+
+ // The server has since recovered. Trigger a segment management tick: the
peon issues an async
+ // re-fetch, whose callback runs on the processing executor and updates
the cached capabilities.
+ peon.loadSegment(segments.get(0), SegmentAction.LOAD, null);
+ recoveringClient.processingExecutor.finishAllPendingTasks();
+
+ // The peon picked up the real turbo capability (3) instead of remaining
pinned to the default (10).
+ Assertions.assertTrue(loadCapabilitiesCalls.get() > 1);
+ Assertions.assertEquals(3,
peon.calculateBatchSize(SegmentLoadingMode.TURBO));
+ }
+
+ @Test
+ public void testOnlyOneCapabilitiesProbeIsOutstandingAtATime()
+ {
+ // A still-unhealthy server should have at most one outstanding capability
probe at a time: while
+ // one is in flight, other doSegmentManagement() ticks must not issue
their own duplicate probes.
+ final AtomicInteger loadCapabilitiesCalls = new AtomicInteger(0);
+ final AlwaysUnhealthyHttpClient unhealthyClient = new
AlwaysUnhealthyHttpClient(loadCapabilitiesCalls);
+
+ final HttpLoadQueuePeon peon = new HttpLoadQueuePeon(
+ "http://dummy:4000";,
+ MAPPER,
+ unhealthyClient,
+ new HttpLoadQueuePeonConfig(null, null, 10),
+ () -> SegmentLoadingMode.NORMAL,
+ new WrappingScheduledExecutorService("HttpLoadQueuePeonTest-%s",
unhealthyClient.processingExecutor, true),
+ unhealthyClient.callbackExecutor
+ );
+
+ // The probe issued during construction failed, so capabilities remain
unconfirmed.
+ Assertions.assertEquals(1, loadCapabilitiesCalls.get());
+
+ // Queue every available segment onto the still-unhealthy peon before
draining anything, i.e.
+ // before any prior probe has had a chance to resolve. Each loadSegment()
call queues exactly one
+ // doSegmentManagement() tick.
+ for (DataSegment segment : segments) {
+ peon.loadSegment(segment, SegmentAction.LOAD, null);
+ }
+
+ // Run exactly those queued ticks, without letting any capability-probe
callback resolve yet (the
+ // callback for the first tick's probe is enqueued behind them, so it
isn't among these).
+ unhealthyClient.processingExecutor.finishNextPendingTasks(segments.size());
+
+ // Only the first of those ticks should have issued a probe; the rest must
see one already
+ // outstanding and skip theirs, instead of each firing its own duplicate
at the struggling server.
+ Assertions.assertEquals(2, loadCapabilitiesCalls.get());
+ }
+
@Test
public void testBatchSize()
{
@@ -350,6 +471,13 @@ public class HttpLoadQueuePeonTest
return success -> httpClient.processedSegments.add(segment);
}
+ private static void respond(HttpResponseHandler<?, ?> handler,
HttpResponseStatus status)
+ {
+ final HttpResponse response = new
DefaultHttpResponse(HttpVersion.HTTP_1_1, status);
+ response.setContent(ChannelBuffers.buffer(0));
+ handler.handleResponse(response, null);
+ }
+
private class TestHttpClient implements HttpClient, DataSegmentChangeHandler
{
final BlockingExecutorService processingExecutor = new
BlockingExecutorService("HttpLoadQueuePeonTest-%s");
@@ -447,6 +575,139 @@ public class HttpLoadQueuePeonTest
}
}
+ /**
+ * An {@link HttpClient} for a server whose loadCapabilities endpoint
returns 503 on the first call
+ * (simulating an unhealthy server at peon construction) and real
capabilities (1, 3) thereafter
+ * (simulating recovery). Segment change requests always succeed.
+ */
+ private static class RecoveringHttpClient implements HttpClient
+ {
+ final BlockingExecutorService processingExecutor = new
BlockingExecutorService("HttpLoadQueuePeonTest-%s");
+ final BlockingExecutorService callbackExecutor = new
BlockingExecutorService("HttpLoadQueuePeonTest-cb");
+ private final AtomicInteger loadCapabilitiesCalls;
+
+ RecoveringHttpClient(AtomicInteger loadCapabilitiesCalls)
+ {
+ this.loadCapabilitiesCalls = loadCapabilitiesCalls;
+ }
+
+ @Override
+ public <Intermediate, Final> ListenableFuture<Final> go(
+ Request request,
+ HttpResponseHandler<Intermediate, Final> handler
+ )
+ {
+ return go(request, handler, null);
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public <Intermediate, Final> ListenableFuture<Final> go(
+ Request request,
+ HttpResponseHandler<Intermediate, Final> handler,
+ Duration duration
+ )
+ {
+ try {
+ if (request.getUrl().toString().contains("/loadCapabilities")) {
+ // First probe (during construction) fails; the server is healthy on
every probe after that.
+ final boolean serverIsHealthy =
loadCapabilitiesCalls.getAndIncrement() > 0;
+ if (serverIsHealthy) {
+ respond(handler, HttpResponseStatus.OK);
+ return (ListenableFuture<Final>) Futures.immediateFuture(
+ new ByteArrayInputStream(
+ MAPPER.writerFor(SegmentLoadingCapabilities.class)
+ .writeValueAsBytes(new SegmentLoadingCapabilities(1,
3))
+ )
+ );
+ }
+ respond(handler, HttpResponseStatus.SERVICE_UNAVAILABLE);
+ return (ListenableFuture<Final>) Futures.immediateFuture(new
ByteArrayInputStream(new byte[0]));
+ }
+
+ // Segment change request: acknowledge every queued segment as
successfully processed.
+ respond(handler, HttpResponseStatus.OK);
+ final List<DataSegmentChangeRequest> changeRequests = MAPPER.readValue(
+ request.getContent().array(),
+ HttpLoadQueuePeon.REQUEST_ENTITY_TYPE_REF
+ );
+ final List<DataSegmentChangeResponse> statuses = new
ArrayList<>(changeRequests.size());
+ for (DataSegmentChangeRequest cr : changeRequests) {
+ statuses.add(new DataSegmentChangeResponse(cr,
SegmentChangeStatus.success()));
+ }
+ return (ListenableFuture<Final>) Futures.immediateFuture(
+ new ByteArrayInputStream(
+
MAPPER.writerFor(HttpLoadQueuePeon.RESPONSE_ENTITY_TYPE_REF).writeValueAsBytes(statuses)
+ )
+ );
+ }
+ catch (Exception ex) {
+ throw new RE(ex, "Unexpected exception.");
+ }
+ }
+ }
+
+ /**
+ * An {@link HttpClient} whose loadCapabilities endpoint always returns 503,
simulating a server
+ * that never recovers. Segment change requests always succeed.
+ */
+ private static class AlwaysUnhealthyHttpClient implements HttpClient
+ {
+ final BlockingExecutorService processingExecutor = new
BlockingExecutorService("HttpLoadQueuePeonTest-%s");
+ final BlockingExecutorService callbackExecutor = new
BlockingExecutorService("HttpLoadQueuePeonTest-cb");
+ private final AtomicInteger loadCapabilitiesCalls;
+
+ AlwaysUnhealthyHttpClient(AtomicInteger loadCapabilitiesCalls)
+ {
+ this.loadCapabilitiesCalls = loadCapabilitiesCalls;
+ }
+
+ @Override
+ public <Intermediate, Final> ListenableFuture<Final> go(
+ Request request,
+ HttpResponseHandler<Intermediate, Final> handler
+ )
+ {
+ return go(request, handler, null);
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public <Intermediate, Final> ListenableFuture<Final> go(
+ Request request,
+ HttpResponseHandler<Intermediate, Final> handler,
+ Duration duration
+ )
+ {
+ try {
+ if (request.getUrl().toString().contains("/loadCapabilities")) {
+ loadCapabilitiesCalls.incrementAndGet();
+ respond(handler, HttpResponseStatus.SERVICE_UNAVAILABLE);
+ return (ListenableFuture<Final>) Futures.immediateFuture(new
ByteArrayInputStream(new byte[0]));
+ }
+
+ // Segment change request: acknowledge every queued segment as
successfully processed.
+ respond(handler, HttpResponseStatus.OK);
+ final List<DataSegmentChangeRequest> changeRequests = MAPPER.readValue(
+ request.getContent().array(),
+ HttpLoadQueuePeon.REQUEST_ENTITY_TYPE_REF
+ );
+ final List<DataSegmentChangeResponse> statuses = new
ArrayList<>(changeRequests.size());
+ for (DataSegmentChangeRequest cr : changeRequests) {
+ statuses.add(new DataSegmentChangeResponse(cr,
SegmentChangeStatus.success()));
+ }
+ return (ListenableFuture<Final>) Futures.immediateFuture(
+ new ByteArrayInputStream(
+
MAPPER.writerFor(HttpLoadQueuePeon.RESPONSE_ENTITY_TYPE_REF).writeValueAsBytes(statuses)
+ )
+ );
+ }
+ catch (Exception ex) {
+ throw new RE(ex, "Unexpected exception.");
+ }
+ }
+ }
+
/**
* Represents an action that can be performed on a segment by calling {@link
#invoke()}.
*/
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
