sklochkov2 commented on code in PR #19754:
URL: https://github.com/apache/druid/pull/19754#discussion_r3659352886
##########
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:
Addressed this in a subsequent commit.
##########
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:
Addressed this in a subsequent commit.
--
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]
