FrankChen021 commented on code in PR #19950:
URL: https://github.com/apache/druid/pull/19950#discussion_r3830377474
##########
server/src/main/java/org/apache/druid/server/coordinator/loading/HttpLoadQueuePeon.java:
##########
@@ -151,48 +170,155 @@ public HttpLoadQueuePeon(
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;
+ }
+ }
- return jsonMapper.readValue(
- stream,
- SegmentLoadingCapabilities.class
+ /**
+ * 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) {
+ 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);
Review Comment:
[P2] Overlapping retries can restore sticky defaults
Refresh requests can overlap. A later successful response may set
capabilitiesConfirmed, then an earlier transient 503 callback can enter this
fallback, restore defaults here without clearing confirmation, and suppress
future refreshes. Serialize refreshes or ignore stale/non-definitive failures
after confirmation, and add an out-of-order response test.
##########
server/src/main/java/org/apache/druid/server/coordinator/loading/HttpLoadQueuePeon.java:
##########
@@ -187,10 +198,23 @@ private SegmentLoadingCapabilities
fetchSegmentLoadingCapabilities()
);
}
catch (Throwable th) {
- throw new RE(th, "Received error while fetching historical capabilities
from Server[%s].", serverId);
+ SegmentLoadingCapabilities defaultCapabilities =
getDefaultLoadingCapabilities();
+ log.makeAlert(
+ th,
+ "Received error while fetching historical capabilities from
Server[%s]. Using default values[%s].",
+ serverId,
+ defaultCapabilities
+ ).emit();
+ return defaultCapabilities;
Review Comment:
Reviewed 3 of 3 changed files. The recovery path is present, but overlapping
probes can still let a late transient failure overwrite confirmed capabilities
and stop future retries. Please add an in-flight guard or only publish
definitive responses, with an out-of-order response test.
<!-- mergelens:review -->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
