FrankChen021 commented on code in PR #19754:
URL: https://github.com/apache/druid/pull/19754#discussion_r3665411511
##########
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:
[P1] Close retained data-server chunks before clearing the queue
`InputStreamHolder.fromByteBuf` now retains every inbound buffer, but
`setupResponseReadFailure` still discards queued holders with `queue.clear()`.
Query timeouts and transport failures therefore drop the only objects capable
of closing those streams after `NettyHttpClient` releases its own inbound
references, leaking pooled direct memory. Drain and close every queued holder
here, as the updated `DirectDruidClient` now does.
##########
server/src/main/java/org/apache/druid/server/lookup/cache/LookupCoordinatorManager.java:
##########
@@ -894,14 +897,56 @@ HttpResponseHandler<InputStream, InputStream>
makeResponseHandler(
final AtomicReference<String> reasonString
)
{
- return new SequenceInputStreamResponseHandler()
+ return new HttpResponseHandler<InputStream, InputStream>()
{
+ private final BytesFullResponseHandler delegate = new
BytesFullResponseHandler();
+ private ClientResponse<BytesFullResponseHolder> delegateResponse;
+
@Override
public ClientResponse<InputStream> handleResponse(HttpResponse
response, TrafficCop trafficCop)
{
- returnCode.set(response.getStatus().getCode());
- reasonString.set(response.getStatus().getReasonPhrase());
- return super.handleResponse(response, trafficCop);
+ returnCode.set(response.status().code());
+ reasonString.set(response.status().reasonPhrase());
+ delegateResponse = delegate.handleResponse(response, trafficCop);
+ return toInputStream(delegateResponse);
+ }
+
+ @Override
+ public ClientResponse<InputStream> handleChunk(
+ ClientResponse<InputStream> response,
+ HttpContent chunk,
+ long chunkNum
+ )
+ {
+ delegateResponse = delegate.handleChunk(delegateResponse, chunk,
chunkNum);
+ return toInputStream(delegateResponse);
+ }
+
+ @Override
+ public ClientResponse<InputStream> done(ClientResponse<InputStream>
response)
+ {
+ delegateResponse = delegate.done(delegateResponse);
+ return toInputStream(delegateResponse);
+ }
+
+ @Override
+ public void exceptionCaught(ClientResponse<InputStream>
clientResponse, Throwable e)
+ {
+ delegate.exceptionCaught(null, e);
+ }
+
+ private ClientResponse<InputStream>
toInputStream(ClientResponse<BytesFullResponseHolder> delegateResponse)
+ {
+ if (delegateResponse == null) {
+ return null;
+ }
+ BytesFullResponseHolder holder = delegateResponse.getObj();
+ InputStream stream = holder == null ? null : new
ByteArrayInputStream(holder.getContent());
Review Comment:
[P1] Avoid copying the accumulated lookup response on every chunk
`toInputStream` is called after every `handleChunk`, and
`holder.getContent()` concatenates every chunk received so far into a new byte
array each time. A response split into n chunks therefore performs quadratic
copying and temporary allocation; large lookup-state responses can cause
extreme GC pressure or OOM. Preserve the previous streaming handler, or
construct the combined byte array only once from `done()`.
##########
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:
Thanks—the retry itself now preserves the bytes, but buffer ownership is
still unresolved. `retainedDuplicate()` lets Netty release only its duplicate,
leaving `Request.content` at `refCnt == 1` after every send. `Request` has no
close/release lifecycle, and `Request.copy()` creates another independently
reference-counted buffer for each Kerberos retry, so pooled/direct request
bodies can remain allocated indefinitely. This also contradicts the existing
`Request#setContent(byte[])` comment that the body is released after the write.
Please store the reusable body in non-reference-counted form and create an
outbound buffer per attempt, or add an explicit lifecycle that releases the
original and retry copies after the final attempt.
Reviewed 167 of 167 files in the supplied full diff.
--
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]
