gianm commented on code in PR #19567:
URL: https://github.com/apache/druid/pull/19567#discussion_r3975177321
##########
processing/src/main/java/org/apache/druid/java/util/http/client/NettyHttpClient.java:
##########
@@ -371,9 +401,13 @@ private void handleExceptionAndCloseChannel(final
Throwable t, final boolean clo
if (!retVal.isDone()) {
if (t instanceof ReadTimeoutException) {
- // ReadTimeoutException thrown by ReadTimeoutHandler is a
singleton with a misleading stack trace.
- // No point including it: instead, we replace it with a fresh
exception.
- retVal.setException(new
ReadTimeoutException(StringUtils.format("[%s] Read timed out", requestDesc)));
+ // ReadTimeoutHandler fires ReadTimeoutException.INSTANCE, a
shared singleton whose stack
+ // trace was captured once at class load and points at
whichever code first touched Netty's
+ // class initializer. Emit a fresh instance so the stack trace
reflects this actual timeout.
+ if (log.isDebugEnabled()) {
+ log.debug("[%s] Read timed out", requestDesc);
+ }
+ retVal.setException(new ReadTimeoutException());
Review Comment:
Hmm. It seems like `ReadTimeoutException` in Netty 4 doesn't have a stack
trace (it inherits a no-op `fillInStackTrace` from `TimeoutException`). I guess
it means we should either stick with the builtin `ReadTimeoutException` from
`ReadTimeoutHandler` or create a new one that has a good message.
If you go with a solution that uses `new ReadTimeoutException()` or
`ReadTimeoutException.INSTANCE` (i.e. no nice message) then at least update the
`HttpResponseHandler` in `DirectDruidClient` to interpolate `e` rather than
`e.getMessage()` into `"Query[%s] url[%s] failed with exception msg [%s]"`.
Maybe only if `e.getMessage()` is null, just to avoid chaanging too many
messages. That way intead of seeing `exception msg [null]` for a timeout it'd
be `exception msg [io.netty.handler.timeout.ReadTimeoutException: null]` (I
think).
##########
processing/src/main/java/org/apache/druid/java/util/http/client/pool/ChannelResourceFactory.java:
##########
@@ -208,62 +222,46 @@ public void operationComplete(ChannelFuture f2)
sslParameters.setEndpointIdentificationAlgorithm("HTTPS");
sslEngine.setSSLParameters(sslParameters);
sslEngine.setUseClientMode(true);
- final SslHandler sslHandler = new SslHandler(
- sslEngine,
- SslHandler.getDefaultBufferPool(),
- false,
- timer,
- sslHandshakeTimeout
- );
+ final SslHandler sslHandler = new SslHandler(sslEngine);
+ sslHandler.setHandshakeTimeoutMillis(sslHandshakeTimeout);
- // https://github.com/netty/netty/issues/160
- sslHandler.setCloseOnSSLException(true);
-
- final ChannelFuture handshakeFuture =
Channels.future(connectFuture.getChannel());
- connectFuture.getChannel().getPipeline().addLast(ERROR_HANDLER_NAME, new
ConnectionErrorHandler(handshakeFuture));
- connectFuture.addListener(
- new ChannelFutureListener()
- {
- @Override
- public void operationComplete(ChannelFuture f)
- {
- if (f.isSuccess()) {
- final ChannelPipeline pipeline = f.getChannel().getPipeline();
- pipeline.addFirst("ssl", sslHandler);
- sslHandler.handshake().addListener(
- new ChannelFutureListener()
- {
- @Override
- public void operationComplete(ChannelFuture f2)
- {
- if (f2.isSuccess()) {
- handshakeFuture.setSuccess();
- } else {
- handshakeFuture.setFailure(
- new ChannelException(
- StringUtils.format("Failed to handshake with
host[%s]", hostname),
- f2.getCause()
- )
- );
- }
- }
- }
- );
- } else {
- handshakeFuture.setFailure(
- new ChannelException(
- StringUtils.format("Failed to connect to host[%s]",
hostname),
- f.getCause()
- )
- );
- }
- }
- }
+ final Channel sslChannel = connectFuture.channel();
+ final ChannelPromise handshakePromise = sslChannel.newPromise();
+ sslChannel.eventLoop().execute(
+ () -> sslChannel.pipeline().addLast(ERROR_HANDLER_NAME, new
ConnectionErrorHandler(handshakePromise))
);
+ connectFuture.addListener((ChannelFuture f) -> {
+ if (f.isSuccess()) {
+ final ChannelPipeline pipeline = f.channel().pipeline();
+ pipeline.addFirst("ssl", sslHandler);
+ sslHandler.handshakeFuture().addListener(f2 -> {
+ if (f2.isSuccess()) {
+ handshakePromise.setSuccess();
Review Comment:
This `setSuccess` and the following two `setFailure` should I think all be
`trySuccess` / `tryFailure` to avoid blowing up on an already-completed promise.
##########
processing/src/main/java/org/apache/druid/java/util/http/client/NettyHttpClient.java:
##########
@@ -125,49 +119,28 @@ public <Intermediate, Final> ListenableFuture<Final> go(
final Channel channel;
final String hostKey = getPoolKey(url);
final ResourceContainer<ChannelFuture> channelResourceContainer =
pool.take(hostKey);
+ // ResourcePool.take is documented as blocking, but a null return is
possible on pool exhaustion or shutdown;
+ // fail fast rather than NPE on the awaitUninterruptibly() below.
+ if (channelResourceContainer == null) {
+ return Futures.immediateFailedFuture(
+ new ChannelException(StringUtils.format("Connection pool exhausted
or timed out for host[%s]", hostKey))
Review Comment:
This message is inaccurate: `pool.take(hostKey)` is only null if `pool` is
closed. (Which shouldn't happen unless the client itself is closed, and would
lead to an `ERROR` being logged anyway.)
Although, I think the container can have a null inside, i.e.,
`channelResourceContainer.get()` can be null in some other cases, such as an
interrupt happening on the thread. It may be worth checking for that (one line
down).
##########
processing/src/main/java/org/apache/druid/java/util/http/client/NettyHttpClient.java:
##########
@@ -371,9 +401,13 @@ private void handleExceptionAndCloseChannel(final
Throwable t, final boolean clo
if (!retVal.isDone()) {
if (t instanceof ReadTimeoutException) {
- // ReadTimeoutException thrown by ReadTimeoutHandler is a
singleton with a misleading stack trace.
- // No point including it: instead, we replace it with a fresh
exception.
- retVal.setException(new
ReadTimeoutException(StringUtils.format("[%s] Read timed out", requestDesc)));
+ // ReadTimeoutHandler fires ReadTimeoutException.INSTANCE, a
shared singleton whose stack
+ // trace was captured once at class load and points at
whichever code first touched Netty's
+ // class initializer. Emit a fresh instance so the stack trace
reflects this actual timeout.
+ if (log.isDebugEnabled()) {
+ log.debug("[%s] Read timed out", requestDesc);
Review Comment:
There is no need to use `log.isDebugEnabled()` here, it's only valuable if
it saves computation. The same check is in `log.debug`.
--
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]
