kfaraz commented on code in PR #20784:
URL: https://github.com/apache/druid/pull/20784#discussion_r4424592918
##########
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:
Good point. The lambda will now be invoked only if the request is not
already completed.
--
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]
