FrankChen021 commented on code in PR #19567:
URL: https://github.com/apache/druid/pull/19567#discussion_r3969865711
##########
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:
[P2] Close the channel when CONNECT times out
When this newly scheduled timeout wins, it only fails
`overallConnectPromise`; it never closes `channel`. `NettyHttpClient.go()` then
returns the failed future to `ResourcePool`, whose `giveBack` retains it and
only calls `factory.close` on a later `take()` for that key or pool shutdown.
Repeated silent or unresponsive proxy connections for different pool keys can
therefore leave open sockets until shutdown. Close the channel on this failure
path and keep the completion handling idempotent.
--
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]
