FrankChen021 commented on code in PR #19754:
URL: https://github.com/apache/druid/pull/19754#discussion_r3657347021
##########
processing/src/main/java/org/apache/druid/java/util/http/client/NettyHttpClient.java:
##########
@@ -125,35 +129,46 @@ public <Intermediate, Final> ListenableFuture<Final> go(
final Channel channel;
final String hostKey = getPoolKey(url);
final ResourceContainer<ChannelFuture> channelResourceContainer =
pool.take(hostKey);
+
+ // Handle pool exhaustion - take() returns null if pool is exhausted or
timed out
+ if (channelResourceContainer == null) {
+ return Futures.immediateFailedFuture(
+ new ChannelException(
+ "Connection pool exhausted or timed out for host: " + hostKey
+ )
+ );
+ }
+
final ChannelFuture channelFuture =
channelResourceContainer.get().awaitUninterruptibly();
if (!channelFuture.isSuccess()) {
channelResourceContainer.returnResource(); // Some other poor sap will
have to deal with it...
return Futures.immediateFailedFuture(
new ChannelException(
"Faulty channel in resource pool",
- channelFuture.getCause()
+ channelFuture.cause()
)
);
} else {
- channel = channelFuture.getChannel();
+ channel = channelFuture.channel();
// In case we get a channel that never had its readability turned back
on.
- channel.setReadable(true);
+ channel.config().setAutoRead(true);
}
final String urlFile =
StringUtils.nullToEmptyNonDruidDataString(url.getFile());
- final HttpRequest httpRequest = new DefaultHttpRequest(
+ final DefaultFullHttpRequest httpRequest = new DefaultFullHttpRequest(
HttpVersion.HTTP_1_1,
method,
- urlFile.isEmpty() ? "/" : urlFile
+ urlFile.isEmpty() ? "/" : urlFile,
+ request.hasContent() ? request.getContent() : Unpooled.EMPTY_BUFFER
Review Comment:
[P1] Preserve request bodies for Kerberos retries
`DefaultFullHttpRequest` takes ownership of this buffer and Netty releases
it after encoding. A subsequent Kerberos 401 retry calls `request.copy()` on
the released buffer, causing `IllegalReferenceCountException` for requests with
bodies. Use independent outbound content or non-reference-counted body storage.
##########
server/src/main/java/org/apache/druid/client/InputStreamHolder.java:
##########
@@ -42,10 +42,11 @@ public static InputStreamHolder fromStream(final
InputStream stream, final long
return new InputStreamHolder(stream, chunkNum, length);
}
- public static InputStreamHolder fromChannelBuffer(final ChannelBuffer
buffer, final long chunkNum)
+ public static InputStreamHolder fromByteBuf(final ByteBuf buffer, final long
chunkNum)
{
final int length = buffer.readableBytes();
- return new InputStreamHolder(new ChannelBufferInputStream(buffer),
chunkNum, length);
+ buffer.retain();
Review Comment:
[P1] Release retained chunks when abandoning streams
`fromByteBuf` retains every buffer, but query close/error paths discard
queued holders with `queue.clear()` without closing them. Repeated cancelled or
failed queries can therefore exhaust pooled direct memory. Drain and close
queued and current streams before clearing them.
--
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]
