jffwang
/
hetu-core
0
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

!994 [I3Y11I] Fix issue in ExchangeClient threads 

Merge pull request !994 from sandygao/I3Y11I-exchange-client-thread
This commit is contained in:
i-robot 2021-06-28 03:37:42 +00:00 committed by Gitee
parent 42f60131ed 1e2ef75f2c
commit 5ca64eae6f
5 changed files with 20 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
presto-main/src/main/java/io/prestosql/execution/SqlStageExecution.java
View File

@ -33,6 +33,7 @@ import io.prestosql.failuredetector.FailureDetector;
import io.prestosql.metadata.InternalNode;
import io.prestosql.metadata.Split;
import io.prestosql.operator.HttpPageBufferClient;
import io.prestosql.server.remotetask.SimpleHttpResponseHandler;
import io.prestosql.snapshot.QuerySnapshotManager;
import io.prestosql.spi.PrestoException;
import io.prestosql.spi.QueryId;
@ -610,6 +611,12 @@ public final class SqlStageExecution
return;
}
}
else if (message.contains(SimpleHttpResponseHandler.EXPECT_200_SAW_5XX)) {
// SimpleHttpResponseHandler can also produce errors that are resumable
log.debug(failure, "Task %s on node %s failed but is resumable. Triggering rescheduling.", taskStatus.getTaskId(), taskStatus.getNodeId());
stateMachine.transitionToResumableFailure();
return;
}
}
stateMachine.transitionToFailed(failure);
}

8
presto-main/src/main/java/io/prestosql/operator/ExchangeClient.java
View File

@ -88,9 +88,11 @@ public class ExchangeClient
private boolean noMoreTargets;
private final Set<String> allTargets = new HashSet<>();
// Markers received before all targets are known. These markers will be sent to all new targets.
private final List<SerializedPage> pendingMarkers = Collections.synchronizedList(new ArrayList<>());
@GuardedBy("this")
private final List<SerializedPage> pendingMarkers = new ArrayList<>();
// pendingOrigins keeps track of the origins in pendingMarkers
private final List<Optional<String>> pendingOrigins = Collections.synchronizedList(new ArrayList<>());
@GuardedBy("this")
private final List<Optional<String>> pendingOrigins = new ArrayList<>();
@GuardedBy("this")
private final Deque<HttpPageBufferClient> queuedClients = new LinkedList<>();
@ -209,7 +211,7 @@ public class ExchangeClient
}
}
public void noMoreTargets()
public synchronized void noMoreTargets()
{
noMoreTargets = true;
pendingMarkers.clear();

2
presto-main/src/main/java/io/prestosql/server/remotetask/HttpRemoteTask.java
View File

@ -103,6 +103,7 @@ import static io.prestosql.protocol.JsonCodecWrapper.unwrapJsonCodec;
import static io.prestosql.protocol.RequestHelpers.setContentTypeHeaders;
import static io.prestosql.server.remotetask.RequestErrorTracker.logError;
import static io.prestosql.spi.StandardErrorCode.REMOTE_HOST_GONE;
import static io.prestosql.spi.StandardErrorCode.REMOTE_TASK_ERROR;
import static io.prestosql.spi.StandardErrorCode.REMOTE_TASK_MISMATCH;
import static io.prestosql.spi.StandardErrorCode.TOO_MANY_REQUESTS_FAILED;
import static io.prestosql.util.Failures.WORKER_NODE_ERROR;
@ -831,6 +832,7 @@ public final class HttpRemoteTask
return failureInfo.getErrorCode().equals(TOO_MANY_REQUESTS_FAILED.toErrorCode())
|| failureInfo.getErrorCode().equals(REMOTE_HOST_GONE.toErrorCode())
|| failureInfo.getErrorCode().equals(REMOTE_TASK_MISMATCH.toErrorCode())
|| failureInfo.getErrorCode().equals(REMOTE_TASK_ERROR.toErrorCode())
|| failureInfo.getMessage() != null && failureInfo.getMessage().contains(WORKER_NODE_ERROR);
}

5
presto-main/src/main/java/io/prestosql/server/remotetask/SimpleHttpResponseHandler.java
View File

@ -30,6 +30,8 @@ import static java.util.Objects.requireNonNull;
public class SimpleHttpResponseHandler<T>
implements FutureCallback<BaseResponse<T>>
{
// Used by SqlStageExecution to detect 5xx failures
public static final CharSequence EXPECT_200_SAW_5XX = format("to be %d, but was 5", HttpStatus.OK.code());
private final SimpleHttpResponseCallback<T> callback;
private final URI uri;
@ -62,6 +64,7 @@ public class SimpleHttpResponseHandler<T>
cause = new PrestoException(REMOTE_TASK_ERROR, format("Expected response from %s is empty", uri));
}
else {
// CAUTION: keep error message consistent with EXPECT_200_SAW_5XX.
cause = new PrestoException(REMOTE_TASK_ERROR, format("Expected response code from %s to be %s, but was %s%n%s",
uri,
HttpStatus.OK.code(),
@ -84,12 +87,14 @@ public class SimpleHttpResponseHandler<T>
private String createErrorMessage(BaseResponse<T> response)
{
if (response instanceof JsonResponseWrapper) {
// CAUTION: keep error message consistent with EXPECT_200_SAW_5XX.
return format("Expected response code from %s to be %s, but was %s%n%s",
uri,
OK.code(),
response.getStatusCode(),
unwrapJsonResponse(response).getResponseBody());
}
// CAUTION: keep error message consistent with EXPECT_200_SAW_5XX.
return format("Expected response code from %s to be %s, but was %s: %s",
uri,
OK.code(),

3
presto-spi/src/main/java/io/prestosql/spi/snapshot/MarkerPage.java
View File

@ -14,7 +14,6 @@
*/
package io.prestosql.spi.snapshot;
import com.google.common.primitives.Ints;
import com.google.common.primitives.Longs;
import io.prestosql.spi.Page;
@ -90,7 +89,7 @@ public class MarkerPage
public byte[] serialize()
{
final int size = Longs.BYTES + Ints.BYTES + 1;
final int size = Longs.BYTES + 1;
return ByteBuffer.allocate(size)
.putLong(snapshotId)
.put((byte) (isResuming ? 1 : 0))