sklochkov2 commented on code in PR #19754:
URL: https://github.com/apache/druid/pull/19754#discussion_r3676355141
##########
processing/src/main/java/org/apache/druid/java/util/http/client/pool/ChannelResourceFactory.java:
##########
@@ -273,78 +304,81 @@ public void operationComplete(ChannelFuture f2)
@Override
public boolean isGood(ChannelFuture resource)
{
- Channel channel = resource.awaitUninterruptibly().getChannel();
+ Channel channel = resource.channel();
boolean isSuccess = resource.isSuccess();
Review Comment:
Please check the latest commits
##########
processing/src/main/java/org/apache/druid/java/util/http/client/NettyHttpClient.java:
##########
@@ -474,4 +520,124 @@ private String getPoolKey(URL url)
return url.getProtocol() + "://" + url.getHost() + ":"
+ (url.getPort() == -1 ? url.getDefaultPort() : url.getPort());
}
+
+ /**
+ * A read-timeout handler that fires a {@link ReadTimeoutException} down the
pipeline if no inbound message is read
+ * within the configured timeout. It behaves like Netty's {@link
io.netty.handler.timeout.ReadTimeoutHandler} but
+ * drives its timer off a shared {@link Timer} (a {@code HashedWheelTimer}
dedicated thread) rather than the channel's
+ * event loop.
+ *
+ * This avoids a class of problems where the event loop's blocking {@code
epoll_wait}/{@code select} is interrupted by
+ * signals (for example a profiler agent), which can reset the wait and
cause event-loop-scheduled timeouts to be
+ * delayed or never fire (see netty/netty#14368 and netty/netty#16244). The
pre-Netty-4 Druid client scheduled read
+ * timeouts on a {@code HashedWheelTimer} for the same reason.
+ */
+ static class TimerReadTimeoutHandler extends ChannelInboundHandlerAdapter
+ {
+ private final Timer timer;
+ private final long timeoutNanos;
+
+ private volatile long lastReadTimeNanos;
+ private volatile Timeout scheduledTimeout;
+ private volatile boolean destroyed;
+ private boolean timedOut;
+
+ TimerReadTimeoutHandler(Timer timer, long timeoutMillis)
+ {
+ this.timer = Preconditions.checkNotNull(timer, "timer");
+ this.timeoutNanos =
Math.max(TimeUnit.MILLISECONDS.toNanos(timeoutMillis), 1L);
+ }
+
+ @Override
+ public void handlerAdded(ChannelHandlerContext ctx)
+ {
+ // The channel is typically already active (taken from the pool) by the
time this handler is added.
+ if (ctx.channel().isActive()) {
+ initialize(ctx);
+ }
+ }
+
+ @Override
+ public void channelActive(ChannelHandlerContext ctx)
+ {
+ initialize(ctx);
+ ctx.fireChannelActive();
+ }
+
+ @Override
+ public void channelRead(ChannelHandlerContext ctx, Object msg)
+ {
+ lastReadTimeNanos = System.nanoTime();
+ ctx.fireChannelRead(msg);
+ }
+
+ @Override
+ public void handlerRemoved(ChannelHandlerContext ctx)
+ {
+ destroy();
+ }
+
+ @Override
+ public void channelInactive(ChannelHandlerContext ctx)
+ {
+ destroy();
+ ctx.fireChannelInactive();
+ }
+
+ private void initialize(ChannelHandlerContext ctx)
+ {
+ if (destroyed) {
+ return;
+ }
+ lastReadTimeNanos = System.nanoTime();
+ schedule(ctx, timeoutNanos);
+ }
+
+ private void schedule(final ChannelHandlerContext ctx, final long
delayNanos)
+ {
+ if (destroyed) {
+ return;
+ }
+ scheduledTimeout = timer.newTimeout(
+ new TimerTask()
+ {
+ @Override
+ public void run(Timeout t)
+ {
+ if (t.isCancelled() || destroyed || !ctx.channel().isOpen()) {
+ return;
+ }
+
+ final long nextDelayNanos = timeoutNanos - (System.nanoTime() -
lastReadTimeNanos);
+ if (nextDelayNanos <= 0) {
+ // Fire the timeout on the event loop, since pipeline events
must run there.
+ ctx.executor().execute(() -> {
Review Comment:
Please check the latest commits
--
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]
