FrankChen021 commented on code in PR #20784:
URL: https://github.com/apache/druid/pull/20784#discussion_r4412523642
##########
server/src/main/java/org/apache/druid/messages/server/MessageRelayResource.java:
##########
@@ -97,78 +92,38 @@ public Void httpGetMessagesFromOutbox(
@QueryParam("epoch") final Long epoch,
@QueryParam("watermark") final Long watermark,
@Context final HttpServletRequest req
- ) throws IOException
+ )
{
if (epoch == null || watermark == null || clientHost == null ||
clientHost.isEmpty()) {
- AsyncContext asyncContext = req.startAsync();
- HttpServletResponse response = (HttpServletResponse)
asyncContext.getResponse();
- response.sendError(HttpServletResponse.SC_BAD_REQUEST);
- asyncContext.complete();
+ AsyncRequest.completed(
+ req,
+ smileMapper,
+ SmileMediaTypes.APPLICATION_JACKSON_SMILE,
+ AsyncResponse.error(HttpServletResponse.SC_BAD_REQUEST, null)
+ );
return null;
}
- final AtomicBoolean didRespond = new AtomicBoolean();
final ListenableFuture<MessageBatch<MessageType>> batchFuture =
outbox.getMessages(clientHost, epoch, watermark);
- final AsyncContext asyncContext = req.startAsync();
- asyncContext.setTimeout(GET_MESSAGES_TIMEOUT);
- asyncContext.addListener(
- ServletResourceUtils.createAsyncTimeoutListener(event -> {
- if (didRespond.compareAndSet(false, true)) {
- HttpServletResponse response = (HttpServletResponse)
asyncContext.getResponse();
- response.setStatus(HttpServletResponse.SC_NO_CONTENT);
- event.getAsyncContext().complete();
- batchFuture.cancel(true);
- }
- })
- );
// Save these items, since "req" becomes inaccessible in future exception
handlers.
final String remoteAddr = req.getRemoteAddr();
final String requestURI = req.getRequestURI();
- Futures.addCallback(
- batchFuture,
- new FutureCallback<>()
- {
- @Override
- public void onSuccess(MessageBatch<MessageType> result)
- {
- if (didRespond.compareAndSet(false, true)) {
- log.debug("Sending message batch: %s", result);
- try {
- HttpServletResponse response = (HttpServletResponse)
asyncContext.getResponse();
- response.setStatus(HttpServletResponse.SC_OK);
-
response.setContentType(SmileMediaTypes.APPLICATION_JACKSON_SMILE);
- smileMapper.writerFor(batchType)
-
.writeValue(asyncContext.getResponse().getOutputStream(), result);
- response.getOutputStream().close();
- asyncContext.complete();
- }
- catch (Exception e) {
- log.noStackTrace().warn(e, "Could not respond to request
from[%s] to[%s]", remoteAddr, requestURI);
- }
- }
- }
-
- @Override
- public void onFailure(Throwable e)
- {
- if (didRespond.compareAndSet(false, true)) {
- try {
- HttpServletResponse response = (HttpServletResponse)
asyncContext.getResponse();
-
response.sendError(HttpServletResponse.SC_INTERNAL_SERVER_ERROR);
- asyncContext.complete();
- }
- catch (Exception e2) {
- e.addSuppressed(e2);
- }
-
- log.noStackTrace().warn(e, "Request failed from[%s] to[%s]",
remoteAddr, requestURI);
- }
- }
- },
- Execs.directExecutor()
- );
+ AsyncRequest.create(req, batchFuture, GET_MESSAGES_TIMEOUT)
+ .contentType(smileMapper,
SmileMediaTypes.APPLICATION_JACKSON_SMILE)
+ .onTimeout(HttpServletResponse.SC_NO_CONTENT)
+ .onSuccess(
+ result -> {
+ log.debug("Sending message batch: %s", result);
+ return AsyncResponse.of(HttpServletResponse.SC_OK,
result, batchType);
+ }
+ )
+ .onFailure(e -> {
+ log.noStackTrace().warn(e, "Request failed from[%s] to[%s]",
remoteAddr, requestURI);
Review Comment:
[P2] Suppress timeout cancellation as a failure
**Finding:** When no message arrives, AsyncRequest completes the response
with 204 and then calls batchFuture.cancel(true). Guava delivers that
cancellation to this direct-executor onFailure callback, which logs a WARN
before the shared completion guard runs. Every normal 30-second idle poll
therefore emits a false Request failed warning and a duplicate-completion
warning, unlike the previous didRespond guard.
**Suggestion:** Have the shared helper distinguish timeout-triggered
cancellation from genuine failures, or keep this warning behind the one-time
response claim.
--
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]
