aho135 commented on code in PR #19950:
URL: https://github.com/apache/druid/pull/19950#discussion_r3846997301
##########
server/src/main/java/org/apache/druid/server/coordinator/loading/HttpLoadQueuePeon.java:
##########
@@ -151,48 +178,160 @@ 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;
+ }
+ }
+
+ /**
+ * 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();
Review Comment:
I think there is some opportunity to common out some code between
`refetchCapabilitiesIfNeeded` and `fetchSegmentLoadingCapabilities`
--
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]
