gianm commented on code in PR #19567:
URL: https://github.com/apache/druid/pull/19567#discussion_r3973334765
##########
processing/src/main/java/org/apache/druid/java/util/http/client/response/StringFullResponseHolder.java:
##########
@@ -19,21 +19,23 @@
package org.apache.druid.java.util.http.client.response;
-import org.jboss.netty.handler.codec.http.HttpResponse;
+import io.netty.handler.codec.http.HttpResponse;
import java.nio.charset.Charset;
public class StringFullResponseHolder extends FullResponseHolder<String>
{
private final StringBuilder builder;
+ @SuppressWarnings("unused")
public StringFullResponseHolder(
HttpResponse response,
Charset charset
Review Comment:
If we don't need it, it's better to remove `charset` rather than adding
`@SuppressWarnings("unused")`.
##########
processing/src/main/java/org/apache/druid/java/util/http/client/response/StringFullResponseHandler.java:
##########
@@ -57,7 +57,7 @@ public ClientResponse<StringFullResponseHolder> handleChunk(
return ClientResponse.finished(null);
}
- holder.addChunk(chunk.getContent().toString(charset));
+ holder.addChunk(chunk.content().toString(charset));
Review Comment:
This shouldn't be doing `toString` on each chunk individually, because
nothing guarantees that chunks are cut on UTF-8 character boundaries. (If a
chunk is cut in the middle of a character, it will become garbled.) I'm not
sure how this worked in netty3 but it would be good to fix it now.
Similar comment for `StatusResponseHandler`.
In both cases we should be able to use `CharsetDecoder` to do a correct
streaming decode.
##########
processing/src/main/java/org/apache/druid/java/util/http/client/HttpClientInit.java:
##########
@@ -125,37 +92,23 @@ public static SSLContext
sslContextWithTrustedKeyStore(final String keyStorePath
}
}
- private static ClientBootstrap createBootstrap(Lifecycle lifecycle, Timer
timer, int bossPoolSize, int workerPoolSize)
+ private static Bootstrap createBootstrap(Lifecycle lifecycle, int
workerPoolSize)
{
- final NioClientBossPool bossPool = new NioClientBossPool(
- Executors.newCachedThreadPool(
- new ThreadFactoryBuilder()
- .setDaemon(true)
- .setNameFormat("HttpClient-Netty-Boss-%s")
- .build()
- ),
- bossPoolSize,
- timer,
- ThreadNameDeterminer.CURRENT
- );
-
- final NioWorkerPool workerPool = new NioWorkerPool(
- Executors.newCachedThreadPool(
- new ThreadFactoryBuilder()
- .setDaemon(true)
- .setNameFormat("HttpClient-Netty-Worker-%s")
- .build()
- ),
+ final NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup(
workerPoolSize,
- ThreadNameDeterminer.CURRENT
+ new ThreadFactoryBuilder()
+ .setDaemon(true)
+ .setNameFormat("HttpClient-Netty-Worker-%s")
+ .build()
);
- final ClientBootstrap bootstrap = new ClientBootstrap(new
NioClientSocketChannelFactory(bossPool, workerPool));
-
- bootstrap.setOption("keepAlive", true);
- bootstrap.setPipelineFactory(new HttpClientPipelineFactory());
+ final Bootstrap bootstrap = new Bootstrap();
+ bootstrap.group(eventLoopGroup)
+ .channel(NioSocketChannel.class)
+ .handler(new HttpClientPipelineFactory())
+ .option(ChannelOption.SO_KEEPALIVE, true);
Review Comment:
The default `connectTimeoutMillis` (which we aren't setting) changed from
10s to 30s from Netty 3 -> 4. We should set it explicitly to 10s and consider
making it configurable.
##########
processing/src/main/java/org/apache/druid/java/util/http/client/netty/HttpClientPipelineFactory.java:
##########
@@ -19,24 +19,21 @@
package org.apache.druid.java.util.http.client.netty;
-import org.jboss.netty.channel.ChannelPipeline;
-import org.jboss.netty.channel.ChannelPipelineFactory;
-import org.jboss.netty.channel.DefaultChannelPipeline;
-import org.jboss.netty.handler.codec.http.HttpClientCodec;
-import org.jboss.netty.handler.codec.http.HttpContentDecompressor;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelPipeline;
+import io.netty.handler.codec.http.HttpClientCodec;
+import io.netty.handler.codec.http.HttpContentDecompressor;
/**
*/
-public class HttpClientPipelineFactory implements ChannelPipelineFactory
+public class HttpClientPipelineFactory extends ChannelInitializer<Channel>
Review Comment:
Rename to `HttpClientChannelInitializer`
##########
processing/src/main/java/org/apache/druid/java/util/http/client/NettyHttpClient.java:
##########
@@ -125,47 +117,61 @@ public <Intermediate, Final> ListenableFuture<Final> go(
final Channel channel;
final String hostKey = getPoolKey(url);
final ResourceContainer<ChannelFuture> channelResourceContainer =
pool.take(hostKey);
Review Comment:
There's a lot of code between here and the error-handling `try` below, where
an exception would cause the `channelResourceContainer` to never be returned
via `returnResource`. If enough leak then connections will start to hang.
Netty 3 and 4 have some differences here that make this interesting: Netty 4
does additional validations in `new DefaultFullHttpRequest` that Netty 3 does
not do in `new DefaultHttpRequest`, so the odds of seeing an exception in this
span of code are higher with Netty 4.
##########
processing/src/main/java/org/apache/druid/java/util/http/client/pool/ChannelResourceFactory.java:
##########
@@ -123,75 +122,80 @@ public ChannelFuture generate(final String hostname)
);
}
- proxyFuture.addListener(new ChannelFutureListener()
- {
- @Override
- public void operationComplete(ChannelFuture f1)
- {
- if (f1.isSuccess()) {
- final Channel channel = f1.getChannel();
- channel.getPipeline().addLast(
- PROXY_HANDLER_NAME,
- new SimpleChannelUpstreamHandler()
+ proxyFuture.addListener((ChannelFuture f1) -> {
+ if (f1.isSuccess()) {
+ final Channel channel = f1.channel();
+ channel.pipeline().addLast(
+ PROXY_HANDLER_NAME,
+ new SimpleChannelInboundHandler<HttpObject>()
+ {
+ private HttpResponseStatus responseStatus;
+
+ @Override
+ protected void channelRead0(ChannelHandlerContext ctx,
HttpObject msg)
{
- @Override
- public void messageReceived(ChannelHandlerContext ctx,
MessageEvent e)
- {
- Object msg = e.getMessage();
-
- final ChannelPipeline pipeline = ctx.getPipeline();
+ if (msg instanceof HttpResponse) {
+ responseStatus = ((HttpResponse) msg).status();
+ }
+ if (msg instanceof LastHttpContent) {
+ final ChannelPipeline pipeline = ctx.pipeline();
pipeline.remove(PROXY_HANDLER_NAME);
- if (msg instanceof HttpResponse) {
- HttpResponse httpResponse = (HttpResponse) msg;
- if
(HttpResponseStatus.OK.equals(httpResponse.getStatus())) {
- // When the HttpClientCodec sees the CONNECT response
complete, it goes into a "done"
- // mode which makes it just do nothing. Swap it with
a new instance that will cover
- // subsequent requests
- pipeline.replace("codec", "codec", new
HttpClientCodec());
- connectFuture.setSuccess();
- } else {
- connectFuture.setFailure(
- new ChannelException(
- StringUtils.format(
- "Got status[%s] from CONNECT request to
proxy[%s]",
- httpResponse.getStatus(),
- proxyUri
- )
- )
- );
- }
+ if (HttpResponseStatus.OK.equals(responseStatus)) {
+ // When the HttpClientCodec sees the CONNECT response
complete, it goes into a "done"
+ // mode which makes it just do nothing. Swap it with a
new instance that will cover
+ // subsequent requests
+ pipeline.replace("codec", "codec", new
HttpClientCodec());
+ overallConnectPromise.setSuccess();
} else {
- connectFuture.setFailure(new
ChannelException(StringUtils.format(
- "Got message of type[%s], don't know what to do.",
msg.getClass()
- )));
- }
- }
- }
- );
- channel.write(connectRequest).addListener(
- new ChannelFutureListener()
- {
- @Override
- public void operationComplete(ChannelFuture f2)
- {
- if (!f2.isSuccess()) {
- connectFuture.setFailure(
+ overallConnectPromise.setFailure(
new ChannelException(
- StringUtils.format("Problem with CONNECT request
to proxy[%s]", proxyUri), f2.getCause()
+ StringUtils.format(
+ "Got status[%s] from CONNECT request to
proxy[%s]",
+ responseStatus,
+ proxyUri
+ )
)
);
}
}
}
- );
- } else {
- connectFuture.setFailure(
- new ChannelException(
- StringUtils.format("Problem connecting to proxy[%s]",
proxyUri), f1.getCause()
- )
- );
- }
+ }
+ );
+ // Bound the wait for the CONNECT response. Bootstrap's
CONNECT_TIMEOUT_MILLIS only covers the TCP
+ // handshake, and the per-request readTimeout in NettyHttpClient
does not apply yet because this
+ // channel is not in the pool. A proxy that accepts TCP but never
sends a CONNECT response would
+ // otherwise hang overallConnectPromise (and every future waiter on
the pool) indefinitely.
+ final ScheduledFuture<?> connectTimeoutTask =
channel.eventLoop().schedule(
Review Comment:
It looks like the fix to this issue is there now, a few lines down. However
it possibly needs to call `overallConnectPromise.tryFailure` rather than
`overallConnectPromise.setFailure` to make multiple failure signals really OK.
##########
processing/src/main/java/org/apache/druid/java/util/http/client/HttpClientInit.java:
##########
@@ -51,41 +47,13 @@ public class HttpClientInit
public static HttpClient createClient(HttpClientConfig config, Lifecycle
lifecycle)
{
try {
- // We need to use the full constructor in order to set a
ThreadNameDeterminer. The other parameters are taken
- // from the defaults in HashedWheelTimer's other constructors.
- final HashedWheelTimer timer = new HashedWheelTimer(
- new ThreadFactoryBuilder().setDaemon(true)
- .setNameFormat("HttpClient-Timer-%s")
- .build(),
- ThreadNameDeterminer.CURRENT,
- 100,
- TimeUnit.MILLISECONDS,
- 512
- );
- lifecycle.addMaybeStartHandler(
- new Lifecycle.Handler()
- {
- @Override
- public void start()
- {
- timer.start();
- }
-
- @Override
- public void stop()
- {
- timer.stop();
- }
- }
- );
return lifecycle.addMaybeStartManagedInstance(
new NettyHttpClient(
new ResourcePool<>(
new ChannelResourceFactory(
- createBootstrap(lifecycle, timer,
config.getBossPoolSize(), config.getWorkerPoolSize()),
Review Comment:
`HttpClientConfig.getBossPoolSize` is unused after this change, please
remove it.
##########
processing/src/main/java/org/apache/druid/java/util/http/client/NettyHttpClient.java:
##########
@@ -370,13 +393,7 @@ 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.
Review Comment:
Netty 4 still seems to throw `ReadTimeoutException.INSTANCE` so I would keep
this block. It was there to make the timeout exception more meaningful.
##########
processing/src/main/java/org/apache/druid/java/util/http/client/pool/ChannelResourceFactory.java:
##########
@@ -208,62 +222,40 @@ 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
- );
-
- // 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 SslHandler sslHandler = new SslHandler(sslEngine);
+ sslHandler.setHandshakeTimeoutMillis(sslHandshakeTimeout);
+
+ final ChannelPromise handshakePromise =
connectFuture.channel().newPromise();
+ connectFuture.channel().pipeline().addLast(ERROR_HANDLER_NAME, new
ConnectionErrorHandler(handshakePromise));
Review Comment:
Apparently in Netty 4, channel bootstrap from the `ChannelInitializer`
happens in the event loop rather than the calling thread, so to be properly
ordered this handler addition has to happen in the event loop too. You should
be able to make that happen by adding it deferredly in `channel.eventLoop()`.
Make sure removal of the handler is done on the event loop too.
--
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]
