FrankChen021 commented on code in PR #19754:
URL: https://github.com/apache/druid/pull/19754#discussion_r3674251650
##########
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:
[P2] Wait for connection futures before rejecting them
`ResourcePool` calls `isGood` immediately after generating or borrowing a
resource. A still-pending connect, proxy, or TLS future reports `isSuccess() ==
false`, so the pool closes a valid in-flight channel and opens a replacement.
This creates redundant connections and handshakes for each host key. Preserve
the previous completion wait before classifying the future.
##########
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:
[P2] Recheck activity before firing the queued timeout
After the timer observes the deadline, it queues this event-loop task but
never rechecks `lastReadTimeNanos`. If the event loop processes an inbound read
before the queued task, `channelRead` refreshes the timestamp and this task
still closes a valid response. Recompute the remaining delay on the event loop
and reschedule when a read arrived.
##########
server/src/main/java/org/apache/druid/discovery/DataServerResponseHandler.java:
##########
@@ -215,11 +223,11 @@ public void exceptionCaught(ClientResponse<InputStream>
clientResponse, Throwabl
setupResponseReadFailure(msg, e);
}
- private boolean enqueue(ChannelBuffer buffer, long chunkNum) throws
InterruptedException
+ private boolean enqueue(ByteBuf buffer, long chunkNum) throws
InterruptedException
{
// Increment queuedByteCount before queueing the object, so
queuedByteCount is at least as high as
// the actual number of queued bytes at any particular time.
- final InputStreamHolder holder =
InputStreamHolder.fromChannelBuffer(buffer, chunkNum);
+ final InputStreamHolder holder = InputStreamHolder.fromByteBuf(buffer,
chunkNum);
Review Comment:
The drain closes holders that were already queued, but it still races with
enqueue. `setupResponseReadFailure` can run on the consumer thread after
`handleChunk` passes `checkQueryTimeout()` but before `queue.put(holder)` on
the event loop; the drain then sees nothing, the late holder retains the
ByteBuf, and `fail` prevents the consumer from ever dequeuing it. Please
serialize enqueue with failure teardown or recheck failure after put and
close/remove the holder. Reviewed 171 of 171 changed files.
--
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]
