Skip to content

TEZ-4460: Read timed out in shuffle handler - incorrect usage of EMPTY_LAST_CONTENT and channel write #257

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
abstractdog merged 2 commits into from
Feb 28, 2023
Merged

TEZ-4460: Read timed out in shuffle handler - incorrect usage of EMPTY_LAST_CONTENT and channel write #257

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e221142
TEZ-4460: Read timed out in shuffle handler - incorrect usage of EMPT…
abstractdog Dec 9, 2022
e321bb7
comment on LastHttpContent.EMPTY_LAST_CONTENT
abstractdog Feb 16, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
31 changes: 16 additions & 15 deletions tez-plugins/tez-aux-services/src/main/java/org/apache/tez/auxservices/ShuffleHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -129,7 +129,6 @@
import io.netty.handler.codec.TooLongFrameException;
import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.DefaultHttpResponse;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.handler.codec.http.HttpHeaders;
import io.netty.handler.codec.http.HttpObjectAggregator;
Expand Down Expand Up @@ -306,21 +305,28 @@ public ReduceMapFileCount(ReduceContext rc) {

@Override
public void operationComplete(ChannelFuture future) throws Exception {
Channel ch = future.channel();
if (!future.isSuccess()) {
future.channel().close();
ch.close();
return;
}
int waitCount = this.reduceContext.getMapsToWait().decrementAndGet();
if (waitCount == 0) {
LOG.debug("Finished with all map outputs");
Copy link
Contributor

@shameersss1 shameersss1 Feb 15, 2023 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Since this is in the hotpath should we enclose this in
if (LOG.isDebugEnabled()) { } ?

Copy link
Contributor Author

@abstractdog abstractdog Feb 16, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this happens once per every shuffle request I guess, and logging parameters don't include expensive operations, so LOG.isDebugEnabled vs. LOG.debug is mostly a method call vs. method call, I don't feel we need to be extremely cautious in this case

/*
* LastHttpContent.EMPTY_LAST_CONTENT can only be written when there are no remaining maps to send,
* this is the only time we can finish the HTTP response.
*/
ch.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT);
Copy link
Contributor

@shameersss1 shameersss1 Feb 15, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

can we add comment why this is required here.

Copy link
Contributor Author

@abstractdog abstractdog Feb 16, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ack, this is the most important part of this patch, added a commit with a code comment

Copy link
Contributor

@rbalamohan rbalamohan Feb 21, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is this issue mainly due to nettty upgrade (4.x?)

Copy link
Contributor Author

@abstractdog abstractdog Feb 21, 2023 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yes, absolutely, this issue is because of the incorrect usage of netty4 APIs (investigation details are on Jira ticket)
most interestingly, there were no unit tests that showed this issue so far (added one now), which reproduces when we fetch more inputs in the same request: due to this issue, the new UT completely hung, and a real TPCDS query on the cluster became very slow, as composite fetch requests hung and timed out eventually (didn't cause a query failure, just an extremely slow query)

metrics.operationComplete(future);
// Let the idle timer handler close keep-alive connections
if (reduceContext.getKeepAlive()) {
ChannelPipeline pipeline = future.channel().pipeline();
ChannelPipeline pipeline = ch.pipeline();
TimeoutHandler timeoutHandler =
(TimeoutHandler) pipeline.get(TIMEOUT_HANDLER);
timeoutHandler.setEnabledTimeout(true);
} else {
future.channel().close();
ch.close();
}
} else {
SHUFFLE.sendMap(reduceContext);
Expand Down Expand Up @@ -993,12 +999,11 @@ public void channelActive(ChannelHandlerContext ctx)
@Override
public void channelRead(ChannelHandlerContext ctx, Object message)
throws Exception {
FullHttpRequest request = (FullHttpRequest) message;
HttpRequest request = (HttpRequest) message;
handleRequest(ctx, request);
request.release();
}

private void handleRequest(ChannelHandlerContext ctx, FullHttpRequest request)
private void handleRequest(ChannelHandlerContext ctx, HttpRequest request)
throws IOException, Exception {
if (request.getMethod() != GET) {
sendError(ctx, METHOD_NOT_ALLOWED);
Expand Down Expand Up @@ -1123,13 +1128,9 @@ private void handleRequest(ChannelHandlerContext ctx, FullHttpRequest request)
for (int i = 0; i < Math.min(maxSessionOpenFiles, mapIds.size()); i++) {
ChannelFuture nextMap = sendMap(reduceContext);
if(nextMap == null) {
// by this special message flushed, we can make sure the whole response is finished
ch.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT);
return;
}
}
// by this special message flushed, we can make sure the whole response is finished
ch.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT);
}

private boolean isNullOrEmpty(List<String> entries) {
Expand Down Expand Up @@ -1496,7 +1497,7 @@ protected ChannelFuture sendMapOutput(ChannelHandlerContext ctx, Channel ch,
DataOutputBuffer dobRange = new DataOutputBuffer();
// Indicate how many record to be written
WritableUtils.writeVInt(dobRange, reduceRange.getLast() - reduceRange.getFirst() + 1);
ch.write(wrappedBuffer(dobRange.getData(), 0, dobRange.getLength()));
ch.writeAndFlush(wrappedBuffer(dobRange.getData(), 0, dobRange.getLength()));
for (int reduce = reduceRange.getFirst(); reduce <= reduceRange.getLast(); reduce++) {
TezIndexRecord index = outputInfo.getIndex(reduce);
// Records are only valid if they have a non-zero part length
Expand All @@ -1511,7 +1512,7 @@ protected ChannelFuture sendMapOutput(ChannelHandlerContext ctx, Channel ch,
DataOutputBuffer dob = new DataOutputBuffer();
header.write(dob);
// Free the memory needed to store the spill and index records
ch.write(wrappedBuffer(dob.getData(), 0, dob.getLength()));
ch.writeAndFlush(wrappedBuffer(dob.getData(), 0, dob.getLength()));
}
outputInfo.finish();

Expand All @@ -1531,14 +1532,14 @@ protected ChannelFuture sendMapOutput(ChannelHandlerContext ctx, Channel ch,
rangeOffset, rangePartLength, manageOsCache, readaheadLength,
readaheadPool, spillFile.getAbsolutePath(),
shuffleBufferSize, shuffleTransferToAllowed);
writeFuture = ch.write(partition);
writeFuture = ch.writeAndFlush(partition);
} else {
// HTTPS cannot be done with zero copy.
final FadvisedChunkedFile chunk = new FadvisedChunkedFile(spill,
rangeOffset, rangePartLength, sslFileBufferSize,
manageOsCache, readaheadLength, readaheadPool,
spillFile.getAbsolutePath());
writeFuture = ch.write(chunk);
writeFuture = ch.writeAndFlush(chunk);
}
metrics.shuffleConnections.incr();
metrics.shuffleOutputBytes.incr(rangePartLength); // optimistic
Expand Down
Loading