You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2021/08/02 23:43:32 UTC
[GitHub] [spark] Victsm opened a new pull request #33613: [SPARK-36378][SHUFFLE] Minor changes to address a few identified push based shuffle server side inefficiencies.
Victsm opened a new pull request #33613:
URL: https://github.com/apache/spark/pull/33613
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the guideline first in
'core/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
We have run performance evaluations on the version of push-based shuffle committed to upstream so far, and have identified a few places for further improvements:
1. On the client side, in addition to randomizing the order of push requests, further randomize the order of blocks within a push request before pushing them. In our benchmark, this has resulted in a 60%-70% reduction of blocks that fail to be merged due to bock collision (the existing block merge ratio is already pretty good in general, and this further improves it).
2. On the server side, we have noticed that the usage of `String.format`, especially when receiving a block push request, has a much higher overhead compared with string concatenation.
3. On the server side, the usage of `Throwables.getStackTraceAsString` in the `ErrorHandler.shouldRetryError` and `ErrorHandler.shouldLogError` has generated quite some overhead. Switch to a direct checking of the exception's message eliminates this overhead.
### Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
Improve shuffle service efficiency and block merge ratio for push-based shuffle.
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as the documentation fix.
If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
If no, write 'No'.
-->
No
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
-->
Existing unit tests.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
mridulm commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r684290557
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/OneForOneBlockPusher.java
##########
@@ -77,42 +81,55 @@ public OneForOneBlockPusher(
@Override
public void onSuccess(ByteBuffer response) {
- // On receipt of a successful block push
- listener.onBlockPushSuccess(blockId, new NioManagedBuffer(ByteBuffer.allocate(0)));
+ BlockPushReturnCode returnCode =
+ (BlockPushReturnCode) BlockTransferMessage.Decoder.fromByteBuffer(response);
+ // If the return code is not SUCCESS, the server has responded some error code. Handle
+ // the error accordingly.
+ if (returnCode.returnCode != ReturnCode.SUCCESS.id()) {
+ checkAndFailRemainingBlocks(index, new BlockPushNonFatalFailure(
+ BlockPushNonFatalFailure.getReturnCode(returnCode.returnCode)));
Review comment:
Move this as a helper static method into `BlockPushNonFatalFailure` ? It is a repeating pattern to create `BlockPushNonFatalFailure` given a `ReturnCode` id.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] otterc commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
otterc commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r682333147
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -863,8 +884,8 @@ public void onComplete(String streamId) throws IOException {
if (isStaleOrTooLate(appShuffleInfo.shuffles.get(partitionInfo.shuffleId),
partitionInfo.shuffleMergeId, partitionInfo.reduceId)) {
deferredBufs = null;
- throw new RuntimeException(String.format("Block %s is %s", streamId,
- ErrorHandler.BlockPushErrorHandler.TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX));
+ throw new RuntimeException("Block " + streamId + " " +
Review comment:
@Victsm there is still another instance in `RemoteBlockPushResolver` that throws `StaleBlockPushException`
```
throw new StaleBlockPushException(String.format("Rejecting shuffle blocks push request"
+ " for shuffle %s with shuffleMergeId %s for application %s_%s as a higher"
+ " shuffleMergeId %s request is already seen", shuffleId, shuffleMergeId,
appShuffleInfo.appId, appShuffleInfo.attemptId, latestShuffleMergeId));
```
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
mridulm commented on pull request #33613:
URL: https://github.com/apache/spark/pull/33613#issuecomment-894691457
Ok to test
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r682266462
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/OneForOneBlockPusher.java
##########
@@ -77,42 +80,55 @@ public OneForOneBlockPusher(
@Override
public void onSuccess(ByteBuffer response) {
- // On receipt of a successful block push
- listener.onBlockPushSuccess(blockId, new NioManagedBuffer(ByteBuffer.allocate(0)));
+ // If the response is non-empty, the server has responded some error code. Decode
+ // it and handle the error accordingly.
+ if (response.hasRemaining()) {
Review comment:
I saw you mentioned the cost of adding extra data for each success push. But I think it's should be fine since the status code only takes 1 byte. Even with the empty body, the `RpcResponse` itself would take some bytes. 1 byte is small compares to the latter.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on pull request #33613:
URL: https://github.com/apache/spark/pull/33613#issuecomment-896697401
Hm, can this test failure:
```
sbt.ForkMain$ForkError: java.lang.AssertionError: Got a non-empty set [Failed to send RPC RPC 6118450341491480151 to <unknown remote>: io.netty.channel.StacklessClosedChannelException]
at org.junit.Assert.fail(Assert.java:89)
at org.junit.Assert.assertTrue(Assert.java:42)
at org.apache.spark.network.RpcIntegrationSuite.assertErrorAndClosed(RpcIntegrationSuite.java:394)
at org.apache.spark.network.RpcIntegrationSuite.sendRpcWithStreamFailures(RpcIntegrationSuite.java:350)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:59)
at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:56)
at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
at org.junit.runners.ParentRunner$3.evaluate(ParentRunner.java:306)
at org.junit.runners.BlockJUnit4ClassRunner$1.evaluate(BlockJUnit4ClassRunner.java:100)
at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:366)
at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:103)
at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:63)
at org.junit.runners.ParentRunner$4.run(ParentRunner.java:331)
at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:79)
at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:329)
at org.junit.runners.ParentRunner.access$100(ParentRunner.java:66)
at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:293)
at org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
at org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27)
at org.junit.runners.ParentRunner$3.evaluate(ParentRunner.java:306)
at org.junit.runners.ParentRunner.run(ParentRunner.java:413)
at org.junit.runners.Suite.runChild(Suite.java:128)
at org.junit.runners.Suite.runChild(Suite.java:27)
at org.junit.runners.ParentRunner$4.run(ParentRunner.java:331)
at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:79)
at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:329)
at org.junit.runners.ParentRunner.access$100(ParentRunner.java:66)
at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:293)
at org.junit.runners.ParentRunner$3.evaluate(ParentRunner.java:306)
at org.junit.runners.ParentRunner.run(ParentRunner.java:413)
at org.junit.runner.JUnitCore.run(JUnitCore.java:137)
at org.junit.runner.JUnitCore.run(JUnitCore.java:115)
at com.novocode.junit.JUnitRunner$1.execute(JUnitRunner.java:132)
at sbt.ForkMain$Run.lambda$runTest$1(ForkMain.java:413)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
```
relate to this PR? Seems like this test started to fail after this change:
- https://github.com/apache/spark/runs/3297511342
- https://github.com/apache/spark/runs/3299079901
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Victsm commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Minor changes to address a few identified push based shuffle server side inefficiencies.
Posted by GitBox <gi...@apache.org>.
Victsm commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r681840796
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -379,9 +399,10 @@ void deleteExecutorDirs(AppShuffleInfo appShuffleInfo) {
@Override
public StreamCallbackWithID receiveBlockDataAsStream(PushBlockStream msg) {
AppShuffleInfo appShuffleInfo = validateAndGetAppShuffleInfo(msg.appId);
- final String streamId = String.format("%s_%d_%d_%d_%d",
- OneForOneBlockPusher.SHUFFLE_PUSH_BLOCK_PREFIX, msg.shuffleId, msg.shuffleMergeId,
- msg.mapIndex, msg.reduceId);
+ // Use string concatenation here to avoid the overhead with String.format on every
Review comment:
This is dependent on the number of blocks pushed, since this is incurred with every block pushed.
The server has no problem to keep up in our benchmark which pushed 40M blocks, but from flamegraph we could see a noticeable CPU time spent in this String.format (~5%).
For push-based shuffle, it would always be good to reduce the server side load.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #33613:
URL: https://github.com/apache/spark/pull/33613#issuecomment-894706364
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/142191/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on pull request #33613:
URL: https://github.com/apache/spark/pull/33613#issuecomment-897268066
Thanks @Victsm!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #33613:
URL: https://github.com/apache/spark/pull/33613#issuecomment-891405357
Can one of the admins verify this patch?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Victsm commented on pull request #33613: [SPARK-36378][SHUFFLE] Minor changes to address a few identified push based shuffle server side inefficiencies.
Posted by GitBox <gi...@apache.org>.
Victsm commented on pull request #33613:
URL: https://github.com/apache/spark/pull/33613#issuecomment-891405302
cc @Ngone51 @mridulm @otterc @attilapiros @venkata91 @zhouyejoe for review.
A few minor changes that have visible performance improvements in our internal perf evaluation.
If needed, I can include the flame graphs in the jira to show more context about some of these changes.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
mridulm commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r683089989
##########
File path: common/network-common/src/main/java/org/apache/spark/network/server/BlockPushNonFatalFailure.java
##########
@@ -0,0 +1,114 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.network.server;
+
+import java.nio.ByteBuffer;
+
+/**
+ * A special RuntimeException thrown when shuffle service experiences a non-fatal failure
+ * with handling block push requests with push-based shuffle. Due to the best-effort nature
+ * of push-based shuffle, there are cases where the exceptions gets thrown under certain
+ * relatively common cases such as when a pushed block is received after the corresponding
+ * shuffle is merge finalized or when a pushed block experiences merge collision. Under these
+ * scenarios, we throw this special RuntimeException.
+ */
+public class BlockPushNonFatalFailure extends RuntimeException {
+
+ /**
+ * The error code of the failure, encoded as a ByteBuffer to be responded back to the client.
+ * Instead of responding a RPCFailure with the exception stack trace as the payload,
+ * which makes checking the content of the exception very tedious on the client side,
+ * we can respond a proper RPCResponse to make it more robust and efficient. This
+ * field is only set on the shuffle server side when the exception is originally generated.
+ */
+ private ByteBuffer response;
+
+ /**
+ * The error code of the failure. This field is only set on the client side when a
+ * BlockPushNonFatalFailure is recreated from the error code received from the server.
+ */
+ private ReturnCode returnCode;
+
+ public BlockPushNonFatalFailure(ByteBuffer response) {
+ this.response = response;
+ }
+
+ public BlockPushNonFatalFailure(ReturnCode returnCode) {
+ this.returnCode = returnCode;
+ }
+
+ /**
+ * Since this type of exception is used to only convey the error code, we reduce the
+ * exception initialization overhead by skipping filling the stack trace.
+ */
+ @Override
+ public synchronized Throwable fillInStackTrace() {
+ return this;
+ }
+
+ public ByteBuffer getResponse() {
Review comment:
Add `Preconditions.checkNotNull` here and in `getReturnCode` to ensure incorrect usage is flagged.
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalBlockStoreClient.java
##########
@@ -218,7 +218,7 @@ public void getMergedBlockMeta(
public void onSuccess(int numChunks, ManagedBuffer buffer) {
logger.trace("Successfully got merged block meta for shuffleId {} shuffleMergeId {}"
+ " reduceId {}", shuffleId, shuffleMergeId, reduceId);
- listener.onSuccess(shuffleId, reduceId, shuffleMergeId,
+ listener.onSuccess(shuffleId, shuffleMergeId, reduceId,
Review comment:
Nice catch !
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -669,6 +696,11 @@ public String getID() {
return streamId;
}
+ @Override
+ public ByteBuffer getCompletionResponse() {
+ return new BlockPushReturnCode(ReturnCode.SUCCESS.id()).toByteBuffer();
+ }
+
Review comment:
nit:
Given how frequently this will be called - make this into a static final read only bytebuffer we duplicate each time it is required ?
Something like:
```
private static final ByteBuffer SUCCESS_BUFFER =
new BlockPushReturnCode(ReturnCode.SUCCESS.id()).toByteBuffer().asReadOnlyBuffer();
```
and getCompletionResponse -> `SUCCESS_BUFFER.duplicate()`
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -373,9 +386,10 @@ void deleteExecutorDirs(AppShuffleInfo appShuffleInfo) {
@Override
public StreamCallbackWithID receiveBlockDataAsStream(PushBlockStream msg) {
AppShuffleInfo appShuffleInfo = validateAndGetAppShuffleInfo(msg.appId);
- final String streamId = String.format("%s_%d_%d_%d_%d",
- OneForOneBlockPusher.SHUFFLE_PUSH_BLOCK_PREFIX, msg.shuffleId, msg.shuffleMergeId,
- msg.mapIndex, msg.reduceId);
+ // Use string concatenation here to avoid the overhead with String.format on every
+ // pushed block.
+ final String streamId = OneForOneBlockPusher.SHUFFLE_PUSH_BLOCK_PREFIX + "_"
+ + msg.shuffleId + "_" + msg.shuffleMergeId + "_" + msg.mapIndex + "_" + msg.reduceId;
if (appShuffleInfo.attemptId != msg.appAttemptId) {
Review comment:
Move this to after the attemptId check.
##########
File path: common/network-common/src/main/java/org/apache/spark/network/server/BlockPushNonFatalFailure.java
##########
@@ -0,0 +1,114 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.network.server;
+
+import java.nio.ByteBuffer;
+
+/**
+ * A special RuntimeException thrown when shuffle service experiences a non-fatal failure
+ * with handling block push requests with push-based shuffle. Due to the best-effort nature
+ * of push-based shuffle, there are cases where the exceptions gets thrown under certain
+ * relatively common cases such as when a pushed block is received after the corresponding
+ * shuffle is merge finalized or when a pushed block experiences merge collision. Under these
+ * scenarios, we throw this special RuntimeException.
+ */
+public class BlockPushNonFatalFailure extends RuntimeException {
+
+ /**
+ * The error code of the failure, encoded as a ByteBuffer to be responded back to the client.
+ * Instead of responding a RPCFailure with the exception stack trace as the payload,
+ * which makes checking the content of the exception very tedious on the client side,
+ * we can respond a proper RPCResponse to make it more robust and efficient. This
+ * field is only set on the shuffle server side when the exception is originally generated.
+ */
+ private ByteBuffer response;
+
+ /**
+ * The error code of the failure. This field is only set on the client side when a
+ * BlockPushNonFatalFailure is recreated from the error code received from the server.
+ */
+ private ReturnCode returnCode;
+
+ public BlockPushNonFatalFailure(ByteBuffer response) {
+ this.response = response;
+ }
+
+ public BlockPushNonFatalFailure(ReturnCode returnCode) {
+ this.returnCode = returnCode;
+ }
+
+ /**
+ * Since this type of exception is used to only convey the error code, we reduce the
+ * exception initialization overhead by skipping filling the stack trace.
+ */
+ @Override
+ public synchronized Throwable fillInStackTrace() {
+ return this;
+ }
+
+ public ByteBuffer getResponse() {
+ return response;
+ }
+
+ public ReturnCode getReturnCode() {
+ return returnCode;
+ }
+
+ public enum ReturnCode {
+ /**
+ * Indicate the case of a successful merge of a pushed block.
+ */
+ SUCCESS(0),
Review comment:
Add a marker at index `0` for `UNKNOWN(0)` or `INVALID(0)` - to help with potential incompatible evolutions.
Thoughts @Ngone51 ?
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -119,7 +122,20 @@ public RemoteBlockPushResolver(TransportConf conf) {
.maximumWeight(conf.mergedIndexCacheSize())
.weigher((Weigher<File, ShuffleIndexInformation>)(file, indexInfo) -> indexInfo.getSize())
.build(ShuffleIndexInformation::new);
- this.errorHandler = new ErrorHandler.BlockPushErrorHandler();
+ this.errorHandler = createErrorHandler();
+ }
+
+ @VisibleForTesting
+ protected ErrorHandler.BlockPushErrorHandler createErrorHandler() {
+ return new ErrorHandler.BlockPushErrorHandler() {
+ // Explicitly use a shuffle service side error handler for handling exceptions.
+ // BlockPushNonException on the server side only has the response field set. It
+ // might require different handling logic compared with a client side error handler.
+ @Override
+ public boolean shouldLogError(Throwable t) {
+ return !(t instanceof BlockPushNonFatalFailure);
+ }
+ };
Review comment:
nit: Make this a static final field and return same instance always
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -379,9 +399,10 @@ void deleteExecutorDirs(AppShuffleInfo appShuffleInfo) {
@Override
public StreamCallbackWithID receiveBlockDataAsStream(PushBlockStream msg) {
AppShuffleInfo appShuffleInfo = validateAndGetAppShuffleInfo(msg.appId);
- final String streamId = String.format("%s_%d_%d_%d_%d",
- OneForOneBlockPusher.SHUFFLE_PUSH_BLOCK_PREFIX, msg.shuffleId, msg.shuffleMergeId,
- msg.mapIndex, msg.reduceId);
+ // Use string concatenation here to avoid the overhead with String.format on every
Review comment:
+1 to avoid `String.format` for simple patterns (unless it is being lazily evaluated on demand).
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/OneForOneBlockPusher.java
##########
@@ -77,42 +80,55 @@ public OneForOneBlockPusher(
@Override
public void onSuccess(ByteBuffer response) {
- // On receipt of a successful block push
- listener.onBlockPushSuccess(blockId, new NioManagedBuffer(ByteBuffer.allocate(0)));
+ BlockPushReturnCode returnCode =
+ (BlockPushReturnCode) BlockTransferMessage.Decoder.fromByteBuffer(response);
+ // If the return code is not 0, the server has responded some error code. Handle
+ // the error accordingly.
+ if (returnCode.returnCode > 0) {
Review comment:
`ReturnCode.SUCCESS != returnCode.returnCode`
##########
File path: core/src/main/scala/org/apache/spark/shuffle/ShuffleBlockPusher.scala
##########
@@ -242,10 +244,16 @@ private[spark] class ShuffleBlockPusher(conf: SparkConf) extends Logging {
handleResult(PushResult(blockId, exception))
}
}
+ // In addition to randomizing the order of the push requests, further randomize the order
+ // of blocks within the push request to further reduce the likelihood of shuffle server side
+ // collision of pushed blocks. This does not increase the cost of reading unmerged shuffle
+ // files on the executor side, because we are still reading MB-size chunks and only randomize
+ // the in-memory sliced buffers post reading.
+ val blocksToPush = Utils.randomize(blockIds.zip(
+ sliceReqBufferIntoBlockBuffers(request.reqBuffer, request.blocks.map(_._2))))
Review comment:
Move this to a different PR - this is not related to the current change.
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -781,8 +819,10 @@ public void onData(String streamId, ByteBuffer buf) throws IOException {
// to disk as well. This way, we avoid having to buffer the entirety of every blocks in
// memory, while still providing the necessary guarantee.
synchronized (partitionInfo) {
- if (isStaleOrTooLate(appShuffleInfo.shuffles.get(partitionInfo.shuffleId),
- partitionInfo.shuffleMergeId, partitionInfo.reduceId)) {
+ if (isStale(appShuffleInfo.shuffles.get(partitionInfo.shuffleId),
+ partitionInfo.shuffleMergeId) ||
+ isTooLate(appShuffleInfo.shuffles.get(partitionInfo.shuffleId),
+ partitionInfo.reduceId)) {
Review comment:
Fetch `appShuffleInfo.shuffles.get(partitionInfo.shuffleId)` and use same instance for both checks.
This pattern is there in `onComplete` and `onFailure` as well.
##########
File path: pusher.diff
##########
@@ -0,0 +1,351 @@
+diff --git a/core/src/main/scala/org/apache/spark/shuffle/ShuffleBlockPusher.scala b/core/src/main/scala/org/apache/spark/shuffle/ShuffleBlockPusher.scala
Review comment:
Looks like it @zhuqi-lucas - please remove this file @Victsm
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -436,17 +456,16 @@ public StreamCallbackWithID receiveBlockDataAsStream(PushBlockStream msg) {
// getting killed. When this happens, we need to distinguish the duplicate blocks as they
// arrive. More details on this is explained in later comments.
- // Track if the block is received after shuffle merge finalized or from an older
- // shuffleMergeId attempt.
- final boolean isStaleBlockOrTooLate = partitionInfoBeforeCheck == null;
// Check if the given block is already merged by checking the bitmap against the given map
// index
- final AppShufflePartitionInfo partitionInfo = isStaleBlockOrTooLate ? null :
+ final AppShufflePartitionInfo partitionInfo = isStaleBlock || isTooLate ? null :
partitionInfoBeforeCheck.mapTracker.contains(msg.mapIndex) ? null : partitionInfoBeforeCheck;
if (partitionInfo != null) {
return new PushBlockStreamCallback(
this, appShuffleInfo, streamId, partitionInfo, msg.mapIndex);
} else {
+ final boolean finalIsTooLate = isTooLate;
+ final boolean finalIsStaleBlock = isStaleBlock;
Review comment:
```suggestion
final BlockPushNonFatalFailure exception = isTooLate ? createTooLate .. : (isStaleBlock ? createStale .. : null);
```
and `if (null != exception) throw exception` in onComplete handler.
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/protocol/BlockPushReturnCode.java
##########
@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.network.shuffle.protocol;
+
+import java.util.Objects;
+
+import io.netty.buffer.ByteBuf;
+import org.apache.commons.lang3.builder.ToStringBuilder;
+import org.apache.commons.lang3.builder.ToStringStyle;
+
+/**
+ * Error code indicating a non-fatal failure of a block push request.
+ * Due to the best-effort nature of push-based shuffle, these failures
+ * do not impact the completion of the block push process. The list of
+ * such errors is in
+ * {@link org.apache.spark.network.server.BlockPushNonFatalFailure.ReturnCode}.
+ *
+ * @since 3.2.0
+ */
+public class BlockPushReturnCode extends BlockTransferMessage {
+ public final byte returnCode;
+
+ public BlockPushReturnCode(byte returnCode) {
+ this.returnCode = returnCode;
Review comment:
Validate if the returnCode is supported.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #33613:
URL: https://github.com/apache/spark/pull/33613#issuecomment-895467356
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/46748/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Victsm commented on pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
Victsm commented on pull request #33613:
URL: https://github.com/apache/spark/pull/33613#issuecomment-894707454
The test failure does not seem related to this patch.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
mridulm commented on pull request #33613:
URL: https://github.com/apache/spark/pull/33613#issuecomment-897074994
Thanks for the investigation @Victsm, @venkata91 !
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Victsm commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
Victsm commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r683766412
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -164,10 +182,7 @@ private AppShufflePartitionInfo getOrCreateAppShufflePartitionInfo(
// current incoming one
int latestShuffleMergeId = appShuffleMergePartitionsInfo.shuffleMergeId;
if (latestShuffleMergeId > shuffleMergeId) {
- throw new StaleBlockPushException(String.format("Rejecting shuffle blocks push request"
- + " for shuffle %s with shuffleMergeId %s for application %s_%s as a higher"
- + " shuffleMergeId %s request is already seen", shuffleId, shuffleMergeId,
- appShuffleInfo.appId, appShuffleInfo.attemptId, latestShuffleMergeId));
Review comment:
Added exception message into `BlockPushNonFatalFailure`, and let the existing logic in ErrorHandler to handle which level to log these.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Victsm commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
Victsm commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r683650921
##########
File path: common/network-common/src/main/java/org/apache/spark/network/server/TransportRequestHandler.java
##########
@@ -213,7 +213,15 @@ public void onData(String streamId, ByteBuffer buf) throws IOException {
public void onComplete(String streamId) throws IOException {
try {
streamHandler.onComplete(streamId);
- callback.onSuccess(ByteBuffer.allocate(0));
+ callback.onSuccess(streamHandler.getCompletionResponse());
+ } catch (BlockPushNonFatalFailure ex) {
Review comment:
We still need to invoke `streamHandler#onFailure` in case we get a non-success return code, which handles logging and other things.
Catching the `BlockPushNonFatalFailure` inside `onComplete` makes it hard to tell apart a success and an error case.
I feel that returning a `ByteBuffer` in `StreamCallback#getCompletionResponse` is generic enough since `StreamCallback` is used for multiple purposes.
Returning a `BlockPushNonFatalFailure` in `StreamCallback` feels too specific for only 1 use case of this interface.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] venkata91 commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Minor changes to address a few identified push based shuffle server side inefficiencies.
Posted by GitBox <gi...@apache.org>.
venkata91 commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r681410950
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -472,8 +493,8 @@ public void onComplete(String streamId) {
if (isStaleBlockOrTooLate) {
// Throw an exception here so the block data is drained from channel and server
// responds RpcFailure to the client.
- throw new RuntimeException(String.format("Block %s %s", streamId,
- ErrorHandler.BlockPushErrorHandler.TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX));
+ throw new RuntimeException("Block " + streamId + " " +
Review comment:
Just a suggestion, may be we can explore if `MessageFormatter` from slf4j would work here or not as it is used in the logging framework so it must be performant enough and at the same time makes the code easy to read.
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ErrorHandler.java
##########
@@ -105,16 +105,16 @@ public boolean shouldRetryError(Throwable t) {
return false;
}
- String errorStackTrace = Throwables.getStackTraceAsString(t);
// If the block is too late or stale block push, there is no need to retry it
- return !errorStackTrace.contains(TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX);
+ String msg = t.getMessage();
+ return !(msg != null && msg.contains(TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX));
Review comment:
may be move this to a helper function?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Victsm commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
Victsm commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r683650921
##########
File path: common/network-common/src/main/java/org/apache/spark/network/server/TransportRequestHandler.java
##########
@@ -213,7 +213,15 @@ public void onData(String streamId, ByteBuffer buf) throws IOException {
public void onComplete(String streamId) throws IOException {
try {
streamHandler.onComplete(streamId);
- callback.onSuccess(ByteBuffer.allocate(0));
+ callback.onSuccess(streamHandler.getCompletionResponse());
+ } catch (BlockPushNonFatalFailure ex) {
Review comment:
We still need to invoke `streamHandler#onFailure` in case we get a non-success return code, which handles logging and other things.
Catching the `BlockPushNonFatalFailure` inside `onComplete` makes it hard to tell apart a success and an error case.
We could potentially return a `BlockPushNonFatalFailure` in `StreamCallback#getCompletionResponse`.
However, I feel that returning a `ByteBuffer` in `StreamCallback#getCompletionResponse` is generic enough since `StreamCallback` is used for multiple purposes.
Returning a `BlockPushNonFatalFailure` in `StreamCallback` feels too specific for only 1 use case of this interface.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] xkrogen commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
xkrogen commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r686168149
##########
File path: common/network-common/src/main/java/org/apache/spark/network/server/BlockPushNonFatalFailure.java
##########
@@ -0,0 +1,162 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.network.server;
+
+import java.nio.ByteBuffer;
+
+import com.google.common.base.Preconditions;
+
+/**
+ * A special RuntimeException thrown when shuffle service experiences a non-fatal failure
+ * with handling block push requests with push-based shuffle. Due to the best-effort nature
+ * of push-based shuffle, there are cases where the exceptions gets thrown under certain
+ * relatively common cases such as when a pushed block is received after the corresponding
+ * shuffle is merge finalized or when a pushed block experiences merge collision. Under these
+ * scenarios, we throw this special RuntimeException.
+ */
+public class BlockPushNonFatalFailure extends RuntimeException {
+ /**
+ * String constant used for generating exception messages indicating a block to be merged
+ * arrives too late on the server side. When we get a block push failure because of the
+ * block arrives too late, we will not retry pushing the block nor log the exception on
+ * the client side.
+ */
+ public static final String TOO_LATE_BLOCK_PUSH_MESSAGE_SUFFIX =
+ " is received after merged shuffle is finalized";
+
+ /**
+ * String constant used for generating exception messages indicating a block to be merged
+ * is a stale block push in the case of indeterminate stage retries on the server side.
+ * When we get a block push failure because of the block push being stale, we will not
+ * retry pushing the block nor log the exception on the client side.
+ */
+ public static final String STALE_BLOCK_PUSH_MESSAGE_SUFFIX =
+ " is a stale block push from an indeterminate stage retry";
+
+ /**
+ * String constant used for generating exception messages indicating the server couldn't
+ * append a block after all available attempts due to collision with other blocks belonging
+ * to the same shuffle partition. When we get a block push failure because of the block
+ * couldn't be written due to this reason, we will not log the exception on the client side.
+ */
+ public static final String BLOCK_APPEND_COLLISION_MSG_SUFFIX =
+ " experienced merge collision on the server side";
+
+ /**
+ * The error code of the failure, encoded as a ByteBuffer to be responded back to the client.
+ * Instead of responding a RPCFailure with the exception stack trace as the payload,
+ * which makes checking the content of the exception very tedious on the client side,
+ * we can respond a proper RPCResponse to make it more robust and efficient. This
+ * field is only set on the shuffle server side when the exception is originally generated.
+ */
+ private ByteBuffer response;
+
+ /**
+ * The error code of the failure. This field is only set on the client side when a
+ * BlockPushNonFatalFailure is recreated from the error code received from the server.
+ */
+ private ReturnCode returnCode;
+
+ public BlockPushNonFatalFailure(ByteBuffer response, String msg) {
+ super(msg);
+ this.response = response;
+ }
+
+ public BlockPushNonFatalFailure(ReturnCode returnCode, String msg) {
+ super(msg);
+ this.returnCode = returnCode;
+ }
+
+ /**
+ * Since this type of exception is used to only convey the error code, we reduce the
+ * exception initialization overhead by skipping filling the stack trace.
+ */
+ @Override
+ public synchronized Throwable fillInStackTrace() {
+ return this;
+ }
+
+ public ByteBuffer getResponse() {
+ // Ensure we do not invoke this method if response is not set
+ Preconditions.checkNotNull(response);
+ return response;
+ }
+
+ public ReturnCode getReturnCode() {
+ // Ensure we do not invoke this method if returnCode is not set
+ Preconditions.checkNotNull(returnCode);
+ return returnCode;
+ }
+
+ public enum ReturnCode {
+ /**
+ * Indicate the case of a successful merge of a pushed block.
+ */
+ SUCCESS(0),
+ /**
+ * Indicate a block to be merged arrives too late on the server side, i.e. after the
+ * corresponding shuffle has been merge finalized. When the client gets this code, it
+ * will not retry pushing the block.
+ */
+ TOO_LATE_BLOCK_PUSH(1),
+ /**
+ * Indicating the server couldn't append a block after all available attempts due to
+ * collision with other blocks belonging to the same shuffle partition.
+ */
+ BLOCK_APPEND_COLLISION_DETECTED(2),
+ /**
+ * Indicate a block received on the server side is a stale block push in the case of
+ * indeterminate stage retries. When the client receives this code, it will not retry
+ * pushing the block.
+ */
+ STALE_BLOCK_PUSH(3);
+
+ private final byte id;
+
+ ReturnCode(int id) {
+ assert id < 128 : "Cannot have more than 128 block push return code";
+ this.id = (byte) id;
+ }
+
+ public byte id() { return id; }
+ }
+
+ public static ReturnCode getReturnCode(byte id) {
+ switch (id) {
+ case 0: return ReturnCode.SUCCESS;
+ case 1: return ReturnCode.TOO_LATE_BLOCK_PUSH;
+ case 2: return ReturnCode.BLOCK_APPEND_COLLISION_DETECTED;
+ case 3: return ReturnCode.STALE_BLOCK_PUSH;
+ default: throw new IllegalArgumentException("Unknown block push return code: " + id);
+ }
+ }
+
+ public static String getErrorMsg(String blockId, ReturnCode errorCode) {
+ Preconditions.checkArgument(errorCode != ReturnCode.SUCCESS);
+ switch (errorCode) {
+ case STALE_BLOCK_PUSH:
+ return "Block " + blockId + STALE_BLOCK_PUSH_MESSAGE_SUFFIX;
+ case TOO_LATE_BLOCK_PUSH:
+ return "Block " + blockId + TOO_LATE_BLOCK_PUSH_MESSAGE_SUFFIX;
+ case BLOCK_APPEND_COLLISION_DETECTED:
+ return "Block " + blockId + BLOCK_APPEND_COLLISION_MSG_SUFFIX;
Review comment:
if there is a 1:1 mapping from error code to message suffix, can you just move the message suffix to be a field on the enum? Then you can just do something like `errorCode.getMessage(blockId)` here, or `return "Block " + blockId + errorCode.getMessageSuffix()`
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
mridulm commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r684290557
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/OneForOneBlockPusher.java
##########
@@ -77,42 +81,55 @@ public OneForOneBlockPusher(
@Override
public void onSuccess(ByteBuffer response) {
- // On receipt of a successful block push
- listener.onBlockPushSuccess(blockId, new NioManagedBuffer(ByteBuffer.allocate(0)));
+ BlockPushReturnCode returnCode =
+ (BlockPushReturnCode) BlockTransferMessage.Decoder.fromByteBuffer(response);
+ // If the return code is not SUCCESS, the server has responded some error code. Handle
+ // the error accordingly.
+ if (returnCode.returnCode != ReturnCode.SUCCESS.id()) {
+ checkAndFailRemainingBlocks(index, new BlockPushNonFatalFailure(
+ BlockPushNonFatalFailure.getReturnCode(returnCode.returnCode)));
Review comment:
Move this as a helper static method into `BlockPushNonFatalFailure` ? It is a repeating pattern to create `BlockPushNonFatalFailure` given a `ReturnCode` id.
Something like `BlockPushNonFatalFailure.fromReturnCide`
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
mridulm commented on pull request #33613:
URL: https://github.com/apache/spark/pull/33613#issuecomment-896161505
Ok to test
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Victsm commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Minor changes to address a few identified push based shuffle server side inefficiencies.
Posted by GitBox <gi...@apache.org>.
Victsm commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r681848866
##########
File path: common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/ErrorHandlerSuite.java
##########
@@ -31,11 +31,11 @@
@Test
public void testErrorRetry() {
ErrorHandler.BlockPushErrorHandler pushHandler = new ErrorHandler.BlockPushErrorHandler();
- assertFalse(pushHandler.shouldRetryError(new RuntimeException(new IllegalArgumentException(
- ErrorHandler.BlockPushErrorHandler.TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX))));
Review comment:
The reason being, in `TransportResponseHandler#handle` in case of a `RPCFailure`, the server side exception stack trace is used to initialize the RuntimeException which gets passed to all the client side RPCResponseCallback.
So the server side information we need to check is always in the first level exception's message.
So, on the client side, we can switch from using `Throwables.getStrackTraceAsString` to just check the exception's message.
On the server side, RemoteBlockPushResolver is also using ErrorHandler to check if it needs to log certain exceptions.
We need to check both the exception and the cause of the exception there.
Of course, what you mentioned in the other comment makes sense, i.e. assuming the exception structure is fragile.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33613:
URL: https://github.com/apache/spark/pull/33613#issuecomment-894703389
Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/46703/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on pull request #33613:
URL: https://github.com/apache/spark/pull/33613#issuecomment-895680912
LGTM if tests pass.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
mridulm commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r684288582
##########
File path: common/network-common/src/main/java/org/apache/spark/network/server/BlockPushNonFatalFailure.java
##########
@@ -0,0 +1,114 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.network.server;
+
+import java.nio.ByteBuffer;
+
+/**
+ * A special RuntimeException thrown when shuffle service experiences a non-fatal failure
+ * with handling block push requests with push-based shuffle. Due to the best-effort nature
+ * of push-based shuffle, there are cases where the exceptions gets thrown under certain
+ * relatively common cases such as when a pushed block is received after the corresponding
+ * shuffle is merge finalized or when a pushed block experiences merge collision. Under these
+ * scenarios, we throw this special RuntimeException.
+ */
+public class BlockPushNonFatalFailure extends RuntimeException {
+
+ /**
+ * The error code of the failure, encoded as a ByteBuffer to be responded back to the client.
+ * Instead of responding a RPCFailure with the exception stack trace as the payload,
+ * which makes checking the content of the exception very tedious on the client side,
+ * we can respond a proper RPCResponse to make it more robust and efficient. This
+ * field is only set on the shuffle server side when the exception is originally generated.
+ */
+ private ByteBuffer response;
+
+ /**
+ * The error code of the failure. This field is only set on the client side when a
+ * BlockPushNonFatalFailure is recreated from the error code received from the server.
+ */
+ private ReturnCode returnCode;
+
+ public BlockPushNonFatalFailure(ByteBuffer response) {
+ this.response = response;
+ }
+
+ public BlockPushNonFatalFailure(ReturnCode returnCode) {
+ this.returnCode = returnCode;
+ }
+
+ /**
+ * Since this type of exception is used to only convey the error code, we reduce the
+ * exception initialization overhead by skipping filling the stack trace.
+ */
+ @Override
+ public synchronized Throwable fillInStackTrace() {
+ return this;
+ }
+
+ public ByteBuffer getResponse() {
+ return response;
+ }
+
+ public ReturnCode getReturnCode() {
+ return returnCode;
+ }
+
+ public enum ReturnCode {
+ /**
+ * Indicate the case of a successful merge of a pushed block.
+ */
+ SUCCESS(0),
Review comment:
That is the existing behavior in the code @Victsm, I want to see if we can make it more robust to evolution.
##########
File path: common/network-common/src/main/java/org/apache/spark/network/server/BlockPushNonFatalFailure.java
##########
@@ -0,0 +1,114 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.network.server;
+
+import java.nio.ByteBuffer;
+
+/**
+ * A special RuntimeException thrown when shuffle service experiences a non-fatal failure
+ * with handling block push requests with push-based shuffle. Due to the best-effort nature
+ * of push-based shuffle, there are cases where the exceptions gets thrown under certain
+ * relatively common cases such as when a pushed block is received after the corresponding
+ * shuffle is merge finalized or when a pushed block experiences merge collision. Under these
+ * scenarios, we throw this special RuntimeException.
+ */
+public class BlockPushNonFatalFailure extends RuntimeException {
+
+ /**
+ * The error code of the failure, encoded as a ByteBuffer to be responded back to the client.
+ * Instead of responding a RPCFailure with the exception stack trace as the payload,
+ * which makes checking the content of the exception very tedious on the client side,
+ * we can respond a proper RPCResponse to make it more robust and efficient. This
+ * field is only set on the shuffle server side when the exception is originally generated.
+ */
+ private ByteBuffer response;
+
+ /**
+ * The error code of the failure. This field is only set on the client side when a
+ * BlockPushNonFatalFailure is recreated from the error code received from the server.
+ */
+ private ReturnCode returnCode;
+
+ public BlockPushNonFatalFailure(ByteBuffer response) {
+ this.response = response;
+ }
+
+ public BlockPushNonFatalFailure(ReturnCode returnCode) {
+ this.returnCode = returnCode;
+ }
+
+ /**
+ * Since this type of exception is used to only convey the error code, we reduce the
+ * exception initialization overhead by skipping filling the stack trace.
+ */
+ @Override
+ public synchronized Throwable fillInStackTrace() {
+ return this;
+ }
+
+ public ByteBuffer getResponse() {
+ return response;
+ }
+
+ public ReturnCode getReturnCode() {
+ return returnCode;
+ }
+
+ public enum ReturnCode {
+ /**
+ * Indicate the case of a successful merge of a pushed block.
+ */
+ SUCCESS(0),
Review comment:
That is the existing behavior in the code @Victsm, I want to see if we can make it more robust to evolution.
If it is causing too many issues, we can fallback on the current compatibility gaurantees.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] gengliangwang commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Minor changes to address a few identified push based shuffle server side inefficiencies.
Posted by GitBox <gi...@apache.org>.
gengliangwang commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r681872829
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ErrorHandler.java
##########
@@ -105,16 +105,16 @@ public boolean shouldRetryError(Throwable t) {
return false;
}
- String errorStackTrace = Throwables.getStackTraceAsString(t);
// If the block is too late or stale block push, there is no need to retry it
- return !errorStackTrace.contains(TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX);
+ String msg = t.getMessage();
+ return !(msg != null && msg.contains(TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX));
Review comment:
@Victsm I planned to cut RC on Aug 10th.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] otterc commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
otterc commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r682338471
##########
File path: common/network-common/src/main/java/org/apache/spark/network/server/BlockPushNonFatalFailure.java
##########
@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.network.server;
+
+import java.nio.ByteBuffer;
+
+/**
+ * A special RuntimeException thrown when shuffle service experiences a non-fatal failure
+ * with handling block push requests with push-based shuffle. Due to the best-effort nature
+ * of push-based shuffle, there are cases where the exceptions gets thrown under certain
+ * relatively common cases such as when a pushed block is received after the corresponding
+ * shuffle is merge finalized or when a pushed block experiences merge collision. Under these
+ * scenarios, we throw this special RuntimeException.
+ */
+public class BlockPushNonFatalFailure extends RuntimeException {
Review comment:
Nit: Why is this called `BlockPushNonFatalFailure` and the error code called `PushBlockNonFatalErrorCode`?
Can both start with `PushBlock...` or `BlockPush...`?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #33613:
URL: https://github.com/apache/spark/pull/33613#issuecomment-894703394
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/46703/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Victsm commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Minor changes to address a few identified push based shuffle server side inefficiencies.
Posted by GitBox <gi...@apache.org>.
Victsm commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r682205256
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -863,8 +884,8 @@ public void onComplete(String streamId) throws IOException {
if (isStaleOrTooLate(appShuffleInfo.shuffles.get(partitionInfo.shuffleId),
partitionInfo.shuffleMergeId, partitionInfo.reduceId)) {
deferredBufs = null;
- throw new RuntimeException(String.format("Block %s is %s", streamId,
- ErrorHandler.BlockPushErrorHandler.TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX));
+ throw new RuntimeException("Block " + streamId + " " +
Review comment:
Marking as resolved since we no longer build a string here in the latest version.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33613:
URL: https://github.com/apache/spark/pull/33613#issuecomment-894694385
**[Test build #142191 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/142191/testReport)** for PR 33613 at commit [`51841e4`](https://github.com/apache/spark/commit/51841e45972086aa08a2c23f227a779bf8e8daf0).
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Minor changes to address a few identified push based shuffle server side inefficiencies.
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r681528196
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -379,9 +399,10 @@ void deleteExecutorDirs(AppShuffleInfo appShuffleInfo) {
@Override
public StreamCallbackWithID receiveBlockDataAsStream(PushBlockStream msg) {
AppShuffleInfo appShuffleInfo = validateAndGetAppShuffleInfo(msg.appId);
- final String streamId = String.format("%s_%d_%d_%d_%d",
- OneForOneBlockPusher.SHUFFLE_PUSH_BLOCK_PREFIX, msg.shuffleId, msg.shuffleMergeId,
- msg.mapIndex, msg.reduceId);
+ // Use string concatenation here to avoid the overhead with String.format on every
Review comment:
What's the exact overhead? It becomes slower?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] otterc commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
otterc commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r682962892
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -471,9 +488,10 @@ public void onData(String streamId, ByteBuffer buf) {
public void onComplete(String streamId) {
if (isStaleBlockOrTooLate) {
// Throw an exception here so the block data is drained from channel and server
- // responds RpcFailure to the client.
- throw new RuntimeException(String.format("Block %s %s", streamId,
- ErrorHandler.BlockPushErrorHandler.TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX));
+ // responds the error code to the client.
+ throw new BlockPushNonFatalFailure(
+ new PushBlockNonFatalErrorCode(ErrorCode.TOO_LATE_OR_STALE_BLOCK_PUSH.id())
Review comment:
@Victsm that's a good point but we can still replace the usage of `StaleBlockPushException` which is created with a message that uses `String.format` with `BlockPushNonFatalFailure`. We can still catch `BlockPushNonFatalFailure` and based on the `ErrorCode` enum differentiate between stale and too late.
Also, for stale block pushes that are for prior application attempts, isn't closing the channel better? For clients that are pushing blocks from prior application attempt, we just want them to stop pushing all together correct?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Victsm commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
Victsm commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r683867359
##########
File path: common/network-common/src/main/java/org/apache/spark/network/server/BlockPushNonFatalFailure.java
##########
@@ -0,0 +1,114 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.network.server;
+
+import java.nio.ByteBuffer;
+
+/**
+ * A special RuntimeException thrown when shuffle service experiences a non-fatal failure
+ * with handling block push requests with push-based shuffle. Due to the best-effort nature
+ * of push-based shuffle, there are cases where the exceptions gets thrown under certain
+ * relatively common cases such as when a pushed block is received after the corresponding
+ * shuffle is merge finalized or when a pushed block experiences merge collision. Under these
+ * scenarios, we throw this special RuntimeException.
+ */
+public class BlockPushNonFatalFailure extends RuntimeException {
+
+ /**
+ * The error code of the failure, encoded as a ByteBuffer to be responded back to the client.
+ * Instead of responding a RPCFailure with the exception stack trace as the payload,
+ * which makes checking the content of the exception very tedious on the client side,
+ * we can respond a proper RPCResponse to make it more robust and efficient. This
+ * field is only set on the shuffle server side when the exception is originally generated.
+ */
+ private ByteBuffer response;
+
+ /**
+ * The error code of the failure. This field is only set on the client side when a
+ * BlockPushNonFatalFailure is recreated from the error code received from the server.
+ */
+ private ReturnCode returnCode;
+
+ public BlockPushNonFatalFailure(ByteBuffer response) {
+ this.response = response;
+ }
+
+ public BlockPushNonFatalFailure(ReturnCode returnCode) {
+ this.returnCode = returnCode;
+ }
+
+ /**
+ * Since this type of exception is used to only convey the error code, we reduce the
+ * exception initialization overhead by skipping filling the stack trace.
+ */
+ @Override
+ public synchronized Throwable fillInStackTrace() {
+ return this;
+ }
+
+ public ByteBuffer getResponse() {
+ return response;
+ }
+
+ public ReturnCode getReturnCode() {
+ return returnCode;
+ }
+
+ public enum ReturnCode {
+ /**
+ * Indicate the case of a successful merge of a pushed block.
+ */
+ SUCCESS(0),
Review comment:
The client side is right now using `BlockPushNonFatalFailure#getReturnCode` for decoding the return code, which throws `IllegalArgumentException` if seeing an unknown return code.
Would that be sufficient?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Victsm commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
Victsm commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r683907444
##########
File path: common/network-common/src/main/java/org/apache/spark/network/server/TransportRequestHandler.java
##########
@@ -213,7 +213,15 @@ public void onData(String streamId, ByteBuffer buf) throws IOException {
public void onComplete(String streamId) throws IOException {
try {
streamHandler.onComplete(streamId);
- callback.onSuccess(ByteBuffer.allocate(0));
+ callback.onSuccess(streamHandler.getCompletionResponse());
+ } catch (BlockPushNonFatalFailure ex) {
Review comment:
What I meant is that by throwing the exception, it's much easier to tell apart the success case from the failure case.
It's not straightforward to tell the content of the returned ByteBuffer without decoding it.
Even for decoding, it's not convenient to decode an rpc message inside network-common module.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] otterc commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
otterc commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r683558307
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -385,14 +397,24 @@ public StreamCallbackWithID receiveBlockDataAsStream(PushBlockStream msg) {
+ "with the current attempt id %s stored in shuffle service for application %s",
msg.appAttemptId, appShuffleInfo.attemptId, msg.appId));
}
+ // Use string concatenation here to avoid the overhead with String.format on every
+ // pushed block.
+ final String streamId = OneForOneBlockPusher.SHUFFLE_PUSH_BLOCK_PREFIX + "_"
+ + msg.shuffleId + "_" + msg.shuffleMergeId + "_" + msg.mapIndex + "_" + msg.reduceId;
// Retrieve merged shuffle file metadata
AppShufflePartitionInfo partitionInfoBeforeCheck;
+ boolean isStaleBlock = false;
+ boolean isTooLate = false;
try {
partitionInfoBeforeCheck = getOrCreateAppShufflePartitionInfo(appShuffleInfo, msg.shuffleId,
msg.shuffleMergeId, msg.reduceId);
- } catch(StaleBlockPushException sbp) {
+ isTooLate = partitionInfoBeforeCheck == null;
+ } catch(BlockPushNonFatalFailure bpf) {
Review comment:
Referring to the previous conversation
https://github.com/apache/spark/pull/33613#discussion_r682952969
I checked other places in `getOrCreateAppShufflePartitionInfo` where we are throwing `RuntimeException` which will just fail to return a `StreamCallback` and thus closing the connection on the server.
```
throw new RuntimeException(
String.format("Cannot initialize merged shuffle partition for appId %s shuffleId %s "
+ "shuffleMergeId %s reduceId %s", appShuffleInfo.appId, shuffleId, shuffleMergeId,
reduceId), e);
```
This is thrown when the server fails to create a merged data/index/meta file for a partition. It may be due to a temporary glitch on the server. So, in this case do we really want to close the connection? The client could be sending blocks of a different partition to this server as well and using the connection for that?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
mridulm commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r683825920
##########
File path: common/network-common/src/main/java/org/apache/spark/network/server/BlockPushNonFatalFailure.java
##########
@@ -0,0 +1,114 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.network.server;
+
+import java.nio.ByteBuffer;
+
+/**
+ * A special RuntimeException thrown when shuffle service experiences a non-fatal failure
+ * with handling block push requests with push-based shuffle. Due to the best-effort nature
+ * of push-based shuffle, there are cases where the exceptions gets thrown under certain
+ * relatively common cases such as when a pushed block is received after the corresponding
+ * shuffle is merge finalized or when a pushed block experiences merge collision. Under these
+ * scenarios, we throw this special RuntimeException.
+ */
+public class BlockPushNonFatalFailure extends RuntimeException {
+
+ /**
+ * The error code of the failure, encoded as a ByteBuffer to be responded back to the client.
+ * Instead of responding a RPCFailure with the exception stack trace as the payload,
+ * which makes checking the content of the exception very tedious on the client side,
+ * we can respond a proper RPCResponse to make it more robust and efficient. This
+ * field is only set on the shuffle server side when the exception is originally generated.
+ */
+ private ByteBuffer response;
+
+ /**
+ * The error code of the failure. This field is only set on the client side when a
+ * BlockPushNonFatalFailure is recreated from the error code received from the server.
+ */
+ private ReturnCode returnCode;
+
+ public BlockPushNonFatalFailure(ByteBuffer response) {
+ this.response = response;
+ }
+
+ public BlockPushNonFatalFailure(ReturnCode returnCode) {
+ this.returnCode = returnCode;
+ }
+
+ /**
+ * Since this type of exception is used to only convey the error code, we reduce the
+ * exception initialization overhead by skipping filling the stack trace.
+ */
+ @Override
+ public synchronized Throwable fillInStackTrace() {
+ return this;
+ }
+
+ public ByteBuffer getResponse() {
+ return response;
+ }
+
+ public ReturnCode getReturnCode() {
+ return returnCode;
+ }
+
+ public enum ReturnCode {
+ /**
+ * Indicate the case of a successful merge of a pushed block.
+ */
+ SUCCESS(0),
Review comment:
Adding `UNKNOWN` allows for evolution of an enum when both client and server cannot be upgraded in sync: allowing for "older" side to fallback to treating the unsupported value as `UNKNOWN` and handle it as a generic error instead of causing deserialization failure or IndexOutOfBound error (in case we are doing `enum.values[id]` sort of code).
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Victsm commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
Victsm commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r683644741
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -385,14 +397,24 @@ public StreamCallbackWithID receiveBlockDataAsStream(PushBlockStream msg) {
+ "with the current attempt id %s stored in shuffle service for application %s",
msg.appAttemptId, appShuffleInfo.attemptId, msg.appId));
}
+ // Use string concatenation here to avoid the overhead with String.format on every
+ // pushed block.
+ final String streamId = OneForOneBlockPusher.SHUFFLE_PUSH_BLOCK_PREFIX + "_"
+ + msg.shuffleId + "_" + msg.shuffleMergeId + "_" + msg.mapIndex + "_" + msg.reduceId;
// Retrieve merged shuffle file metadata
AppShufflePartitionInfo partitionInfoBeforeCheck;
+ boolean isStaleBlock = false;
+ boolean isTooLate = false;
try {
partitionInfoBeforeCheck = getOrCreateAppShufflePartitionInfo(appShuffleInfo, msg.shuffleId,
msg.shuffleMergeId, msg.reduceId);
- } catch(StaleBlockPushException sbp) {
+ isTooLate = partitionInfoBeforeCheck == null;
+ } catch(BlockPushNonFatalFailure bpf) {
Review comment:
Unable to create these files seems like a more disrupting issue.
It could be a temporary glitch, but it could also be due to issues like disk failures or disk being full, which will just fail all other block pushes even for different partitions.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] zhuqi-lucas commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
zhuqi-lucas commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r682301682
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -471,9 +488,10 @@ public void onData(String streamId, ByteBuffer buf) {
public void onComplete(String streamId) {
if (isStaleBlockOrTooLate) {
// Throw an exception here so the block data is drained from channel and server
- // responds RpcFailure to the client.
- throw new RuntimeException(String.format("Block %s %s", streamId,
- ErrorHandler.BlockPushErrorHandler.TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX));
+ // responds the error code to the client.
+ throw new BlockPushNonFatalFailure(
+ new PushBlockNonFatalErrorCode(ErrorCode.TOO_LATE_OR_STALE_BLOCK_PUSH.id())
Review comment:
I also think we'd better use error codes for "too late push" and "stale push" separately, and after this, i will fill the "too old attempt" case in [SPARK-35548](https://issues.apache.org/jira/browse/SPARK-35548).
##########
File path: pusher.diff
##########
@@ -0,0 +1,351 @@
+diff --git a/core/src/main/scala/org/apache/spark/shuffle/ShuffleBlockPusher.scala b/core/src/main/scala/org/apache/spark/shuffle/ShuffleBlockPusher.scala
Review comment:
@Victsm Is this file pushed by mistake?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #33613:
URL: https://github.com/apache/spark/pull/33613#issuecomment-894703394
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/46703/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33613:
URL: https://github.com/apache/spark/pull/33613#issuecomment-896287055
Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/46797/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Minor changes to address a few identified push based shuffle server side inefficiencies.
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r681529378
##########
File path: common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/ErrorHandlerSuite.java
##########
@@ -31,11 +31,11 @@
@Test
public void testErrorRetry() {
ErrorHandler.BlockPushErrorHandler pushHandler = new ErrorHandler.BlockPushErrorHandler();
- assertFalse(pushHandler.shouldRetryError(new RuntimeException(new IllegalArgumentException(
- ErrorHandler.BlockPushErrorHandler.TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX))));
Review comment:
Why do we strip the `IllegalArgumentException` here?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r683215389
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -164,10 +182,7 @@ private AppShufflePartitionInfo getOrCreateAppShufflePartitionInfo(
// current incoming one
int latestShuffleMergeId = appShuffleMergePartitionsInfo.shuffleMergeId;
if (latestShuffleMergeId > shuffleMergeId) {
- throw new StaleBlockPushException(String.format("Rejecting shuffle blocks push request"
- + " for shuffle %s with shuffleMergeId %s for application %s_%s as a higher"
- + " shuffleMergeId %s request is already seen", shuffleId, shuffleMergeId,
- appShuffleInfo.appId, appShuffleInfo.attemptId, latestShuffleMergeId));
Review comment:
Could we also log for the case of:
```
if (dataFile.exists()) {
return null;
}
```
Btw, is it equivalent to throw `BlockPushNonFatalFailure(ReturnCode.TOO_LATE_BLOCK_PUSH)` rather than return null? If it's equivalent, I think throwing exception is more consistent with the case of "stale push"
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33613:
URL: https://github.com/apache/spark/pull/33613#issuecomment-896323854
**[Test build #142289 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/142289/testReport)** for PR 33613 at commit [`52b7093`](https://github.com/apache/spark/commit/52b709356b367a8616acffec9f7f3a0eec834fce).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #33613:
URL: https://github.com/apache/spark/pull/33613#issuecomment-896162488
**[Test build #142284 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/142284/testReport)** for PR 33613 at commit [`edde291`](https://github.com/apache/spark/commit/edde291ac8458a272279e47ce54cf2c35e31de43).
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r685642507
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -748,20 +780,26 @@ private void incrementIOExceptionsAndAbortIfNecessary() {
}
/**
- * If appShuffleMergePartitionsInfo is null or shuffleMergePartitions is set to
- * INDETERMINATE_SHUFFLE_FINALIZED or if the reduceId is not in the map then the
- * shuffle is already finalized. Therefore the block push is too late. If
- * appShuffleMergePartitionsInfo's shuffleMergeId is
+ * If appShuffleMergePartitionsInfo's shuffleMergeId is
* greater than the request shuffleMergeId then it is a stale block push.
*/
- private boolean isStaleOrTooLate(
+ private boolean isStale(
AppShuffleMergePartitionsInfo appShuffleMergePartitionsInfo,
- int shuffleMergeId,
- int reduceId) {
+ int shuffleMergeId) {
+ return appShuffleMergePartitionsInfo.shuffleMergeId > shuffleMergeId;
+ }
+
+ /**
+ * If appShuffleMergePartitionsInfo is null or shuffleMergePartitions is set to
+ * INDETERMINATE_SHUFFLE_FINALIZED or if the reduceId is not in the map then the
+ * shuffle is already finalized. Therefore the block push is too late.
+ */
+ private boolean isTooLate(
+ AppShuffleMergePartitionsInfo appShuffleMergePartitionsInfo,
+ int reduceId) {
Review comment:
nit: 4 indents
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] venkata91 commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
venkata91 commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r682840984
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -471,9 +488,10 @@ public void onData(String streamId, ByteBuffer buf) {
public void onComplete(String streamId) {
if (isStaleBlockOrTooLate) {
// Throw an exception here so the block data is drained from channel and server
- // responds RpcFailure to the client.
- throw new RuntimeException(String.format("Block %s %s", streamId,
- ErrorHandler.BlockPushErrorHandler.TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX));
+ // responds the error code to the client.
+ throw new BlockPushNonFatalFailure(
+ new PushBlockNonFatalErrorCode(ErrorCode.TOO_LATE_OR_STALE_BLOCK_PUSH.id())
Review comment:
This is used with in `RemoteBlockPushResolver`, we need it to reject older `shuffleMergeId` block push requests. See this https://github.com/apache/spark/blob/d8169493b662acac31d3fc5e6c5051917428c974/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java#L399 This is required since we cannot return null in the `compute` as that would remove the `shuffleId` Also see this comment - https://github.com/apache/spark/pull/33034#discussion_r676098927
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Victsm commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
Victsm commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r682952969
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -471,9 +488,10 @@ public void onData(String streamId, ByteBuffer buf) {
public void onComplete(String streamId) {
if (isStaleBlockOrTooLate) {
// Throw an exception here so the block data is drained from channel and server
- // responds RpcFailure to the client.
- throw new RuntimeException(String.format("Block %s %s", streamId,
- ErrorHandler.BlockPushErrorHandler.TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX));
+ // responds the error code to the client.
+ throw new BlockPushNonFatalFailure(
+ new PushBlockNonFatalErrorCode(ErrorCode.TOO_LATE_OR_STALE_BLOCK_PUSH.id())
Review comment:
@otterc you are right that we can just replace `StaleBlockPushException` with `BlockPushNonFatalFailure`. Catching it and later throwing another is not necessary.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] otterc commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
otterc commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r682962892
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -471,9 +488,10 @@ public void onData(String streamId, ByteBuffer buf) {
public void onComplete(String streamId) {
if (isStaleBlockOrTooLate) {
// Throw an exception here so the block data is drained from channel and server
- // responds RpcFailure to the client.
- throw new RuntimeException(String.format("Block %s %s", streamId,
- ErrorHandler.BlockPushErrorHandler.TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX));
+ // responds the error code to the client.
+ throw new BlockPushNonFatalFailure(
+ new PushBlockNonFatalErrorCode(ErrorCode.TOO_LATE_OR_STALE_BLOCK_PUSH.id())
Review comment:
@Victsm that's a good point. But we can still replace the usage of `StaleBlockPushException` which is created with a message that uses `String.format` with `BlockPushNonFatalFailure`. We can still catch `BlockPushNonFatalFailure` and based on the `ErrorCode` enum differentiate between stale and too late.
Also, for stale block pushes that are for prior application attempts, isn't it closing the channel better? For clients that are pushing blocks from prior application attempt, we just want them to stop pushing all together correct.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33613:
URL: https://github.com/apache/spark/pull/33613#issuecomment-894705920
**[Test build #142191 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/142191/testReport)** for PR 33613 at commit [`51841e4`](https://github.com/apache/spark/commit/51841e45972086aa08a2c23f227a779bf8e8daf0).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
mridulm commented on pull request #33613:
URL: https://github.com/apache/spark/pull/33613#issuecomment-897005267
@Victsm, @venkata91, @zhouyejoe Please take a look at @HyukjinKwon's comment above.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] otterc commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
otterc commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r682876739
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -471,9 +488,10 @@ public void onData(String streamId, ByteBuffer buf) {
public void onComplete(String streamId) {
if (isStaleBlockOrTooLate) {
// Throw an exception here so the block data is drained from channel and server
- // responds RpcFailure to the client.
- throw new RuntimeException(String.format("Block %s %s", streamId,
- ErrorHandler.BlockPushErrorHandler.TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX));
+ // responds the error code to the client.
+ throw new BlockPushNonFatalFailure(
+ new PushBlockNonFatalErrorCode(ErrorCode.TOO_LATE_OR_STALE_BLOCK_PUSH.id())
Review comment:
I saw how it was being used. My point is that this is not needed any more. We can use `BlockPushNoneFatalFailure` instead of `StaleBlockPushException` where it is instantiated.
Also, catching the `StaleBlockPushException` is not really required. I am pointing to the below lines. If `getOrCreateAppShufflePartitionInfo` throws a `RuntimeException`, then `receiveBlockDataAsStream` would just fail. In case of stale push, it's not really needed that a `StreamCallback` is returned.
```
try {
partitionInfoBeforeCheck = getOrCreateAppShufflePartitionInfo(appShuffleInfo, msg.shuffleId,
msg.shuffleMergeId, msg.reduceId);
} catch(StaleBlockPushException sbp) {
// Set partitionInfoBeforeCheck to null so that stale block push gets handled.
partitionInfoBeforeCheck = null;
}
```
So, this could just simplified with
```
AppShufflePartitionInfo partitionInfoBeforeCheck = getOrCreateAppShufflePartitionInfo(appShuffleInfo, msg.shuffleId,
msg.shuffleMergeId, msg.reduceId);
```
If the execution goes beyond this, then that means the push was not stale and then it is just too late.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #33613:
URL: https://github.com/apache/spark/pull/33613#issuecomment-896287094
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r683220812
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -385,14 +397,24 @@ public StreamCallbackWithID receiveBlockDataAsStream(PushBlockStream msg) {
+ "with the current attempt id %s stored in shuffle service for application %s",
msg.appAttemptId, appShuffleInfo.attemptId, msg.appId));
}
+ // Use string concatenation here to avoid the overhead with String.format on every
+ // pushed block.
+ final String streamId = OneForOneBlockPusher.SHUFFLE_PUSH_BLOCK_PREFIX + "_"
+ + msg.shuffleId + "_" + msg.shuffleMergeId + "_" + msg.mapIndex + "_" + msg.reduceId;
// Retrieve merged shuffle file metadata
AppShufflePartitionInfo partitionInfoBeforeCheck;
+ boolean isStaleBlock = false;
+ boolean isTooLate = false;
Review comment:
Why not just use the `ReturnCode` filed directly?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on pull request #33613:
URL: https://github.com/apache/spark/pull/33613#issuecomment-895680790
@Victsm Please resolve the conflicts
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r683905357
##########
File path: common/network-common/src/main/java/org/apache/spark/network/server/TransportRequestHandler.java
##########
@@ -213,7 +213,15 @@ public void onData(String streamId, ByteBuffer buf) throws IOException {
public void onComplete(String streamId) throws IOException {
try {
streamHandler.onComplete(streamId);
- callback.onSuccess(ByteBuffer.allocate(0));
+ callback.onSuccess(streamHandler.getCompletionResponse());
+ } catch (BlockPushNonFatalFailure ex) {
Review comment:
The current `getCompletionResponse` only returns the `SUCCESS` byte buffer, right?
I didn't mean to return `BlockPushNonFatalFailure` in the `StreamCallback. getCompletionResponse`. What I expect is to allow return `SUCCESS`, `TOO_LATE_BLOCK_PUSH`, etc, byte buffers in `getCompletionResponse`.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33613:
URL: https://github.com/apache/spark/pull/33613#issuecomment-896214847
**[Test build #142289 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/142289/testReport)** for PR 33613 at commit [`52b7093`](https://github.com/apache/spark/commit/52b709356b367a8616acffec9f7f3a0eec834fce).
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33613:
URL: https://github.com/apache/spark/pull/33613#issuecomment-896261793
Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/46797/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon edited a comment on pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
HyukjinKwon edited a comment on pull request #33613:
URL: https://github.com/apache/spark/pull/33613#issuecomment-897268066
Thanks @mridulm, @Victsm and @venkata91!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] venkata91 commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
venkata91 commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r682310400
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -471,9 +488,10 @@ public void onData(String streamId, ByteBuffer buf) {
public void onComplete(String streamId) {
if (isStaleBlockOrTooLate) {
// Throw an exception here so the block data is drained from channel and server
- // responds RpcFailure to the client.
- throw new RuntimeException(String.format("Block %s %s", streamId,
- ErrorHandler.BlockPushErrorHandler.TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX));
+ // responds the error code to the client.
+ throw new BlockPushNonFatalFailure(
+ new PushBlockNonFatalErrorCode(ErrorCode.TOO_LATE_OR_STALE_BLOCK_PUSH.id())
Review comment:
May be it is possible with the introduction of `StaleBlockPushException`.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r683215389
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -164,10 +182,7 @@ private AppShufflePartitionInfo getOrCreateAppShufflePartitionInfo(
// current incoming one
int latestShuffleMergeId = appShuffleMergePartitionsInfo.shuffleMergeId;
if (latestShuffleMergeId > shuffleMergeId) {
- throw new StaleBlockPushException(String.format("Rejecting shuffle blocks push request"
- + " for shuffle %s with shuffleMergeId %s for application %s_%s as a higher"
- + " shuffleMergeId %s request is already seen", shuffleId, shuffleMergeId,
- appShuffleInfo.appId, appShuffleInfo.attemptId, latestShuffleMergeId));
Review comment:
Could we also log for the case of:
```
if (dataFile.exists()) {
return null;
}
```
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Victsm commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
Victsm commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r682988604
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -471,9 +488,10 @@ public void onData(String streamId, ByteBuffer buf) {
public void onComplete(String streamId) {
if (isStaleBlockOrTooLate) {
// Throw an exception here so the block data is drained from channel and server
- // responds RpcFailure to the client.
- throw new RuntimeException(String.format("Block %s %s", streamId,
- ErrorHandler.BlockPushErrorHandler.TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX));
+ // responds the error code to the client.
+ throw new BlockPushNonFatalFailure(
+ new PushBlockNonFatalErrorCode(ErrorCode.TOO_LATE_OR_STALE_BLOCK_PUSH.id())
Review comment:
For stale push from prior app attempt, it currently indeed throws an IllegalArgumentException inside `receiveBlockDataAsStream` which would lead to server closing channels.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Victsm commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
Victsm commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r683646140
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/OneForOneBlockPusher.java
##########
@@ -77,42 +81,55 @@ public OneForOneBlockPusher(
@Override
public void onSuccess(ByteBuffer response) {
- // On receipt of a successful block push
- listener.onBlockPushSuccess(blockId, new NioManagedBuffer(ByteBuffer.allocate(0)));
+ BlockPushReturnCode returnCode =
+ (BlockPushReturnCode) BlockTransferMessage.Decoder.fromByteBuffer(response);
+ // If the return code is not SUCCESS, the server has responded some error code. Handle
+ // the error accordingly.
+ if (returnCode.returnCode != ReturnCode.SUCCESS.id()) {
+ checkAndFailRemainingBlocks(index, new BlockPushNonFatalFailure(
+ BlockPushNonFatalFailure.getReturnCode(returnCode.returnCode)));
+ } else {
+ // On receipt of a successful block push
+ listener.onBlockPushSuccess(blockId, new NioManagedBuffer(ByteBuffer.allocate(0)));
+ }
}
@Override
public void onFailure(Throwable e) {
- // Since block push is best effort, i.e., if we encounter a block push failure that's still
- // retriable according to ErrorHandler (not a connection exception and the block is not too
- // late), we should not fail all remaining block pushes even though
- // RetryingBlockTransferor might consider this failure not retriable (exceeding max retry
- // count etc). The best effort nature makes block push tolerable of a partial completion.
- // Thus, we only fail the block that's actually failed in this case. Note that, on the
- // RetryingBlockTransferor side, if retry is initiated, it would still invalidate the
- // previous active retry listener, and retry pushing all outstanding blocks. However, since
- // the blocks to be pushed are preloaded into memory and the first attempt of pushing these
- // blocks might have already succeeded, retry pushing all the outstanding blocks should be
- // very cheap (on the client side, the block data is in memory; on the server side, the block
- // will be recognized as a duplicate which triggers noop handling). Here, by failing only the
- // one block that's actually failed, we are essentially preventing forwarding unnecessary
- // block push failures to the parent listener of the retry listener.
- //
- // Take the following as an example. For the common exception during block push handling,
- // i.e. block collision, it is considered as retriable by ErrorHandler but not retriable
- // by RetryingBlockTransferor. When we encounter a failure of this type, we only fail the
- // one block encountering this issue not the remaining blocks in the same batch. On the
- // RetryingBlockTransferor side, since this exception is considered as not retriable, it
- // would immediately invoke parent listener's onBlockTransferFailure. However, the remaining
- // blocks in the same batch would remain current and active and they won't be impacted by
- // this exception.
- if (PUSH_ERROR_HANDLER.shouldRetryError(e)) {
- String[] targetBlockId = Arrays.copyOfRange(blockIds, index, index + 1);
- failRemainingBlocks(targetBlockId, e);
- } else {
- String[] targetBlockId = Arrays.copyOfRange(blockIds, index, blockIds.length);
- failRemainingBlocks(targetBlockId, e);
- }
+ checkAndFailRemainingBlocks(index, e);
+ }
+ }
+
+ private void checkAndFailRemainingBlocks(int index, Throwable e) {
+ // Since block push is best effort, i.e., if we encounter a block push failure that's still
+ // retriable according to ErrorHandler (not a connection exception and the block is not too
+ // late), we should not fail all remaining block pushes even though
+ // RetryingBlockTransferor might consider this failure not retriable (exceeding max retry
+ // count etc). The best effort nature makes block push tolerable of a partial completion.
+ // Thus, we only fail the block that's actually failed in this case. Note that, on the
+ // RetryingBlockTransferor side, if retry is initiated, it would still invalidate the
+ // previous active retry listener, and retry pushing all outstanding blocks. However, since
+ // the blocks to be pushed are preloaded into memory and the first attempt of pushing these
+ // blocks might have already succeeded, retry pushing all the outstanding blocks should be
+ // very cheap (on the client side, the block data is in memory; on the server side, the block
+ // will be recognized as a duplicate which triggers noop handling). Here, by failing only the
+ // one block that's actually failed, we are essentially preventing forwarding unnecessary
+ // block push failures to the parent listener of the retry listener.
+ //
+ // Take the following as an example. For the common exception during block push handling,
+ // i.e. block collision, it is considered as retriable by ErrorHandler but not retriable
+ // by RetryingBlockTransferor. When we encounter a failure of this type, we only fail the
+ // one block encountering this issue not the remaining blocks in the same batch. On the
+ // RetryingBlockTransferor side, since this exception is considered as not retriable, it
+ // would immediately invoke parent listener's onBlockTransferFailure. However, the remaining
+ // blocks in the same batch would remain current and active and they won't be impacted by
+ // this exception.
Review comment:
We have the existing handling in `ErrorHandler#shouldLogError` to decide whether to log the error or not.
Certain issues such as too late block push or block push collision can be more frequent, and logging at info level for every occurrence of these issues on either server side or client side could pollute the logs.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
mridulm commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r683826368
##########
File path: common/network-common/src/main/java/org/apache/spark/network/server/BlockPushNonFatalFailure.java
##########
@@ -0,0 +1,114 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.network.server;
+
+import java.nio.ByteBuffer;
+
+/**
+ * A special RuntimeException thrown when shuffle service experiences a non-fatal failure
+ * with handling block push requests with push-based shuffle. Due to the best-effort nature
+ * of push-based shuffle, there are cases where the exceptions gets thrown under certain
+ * relatively common cases such as when a pushed block is received after the corresponding
+ * shuffle is merge finalized or when a pushed block experiences merge collision. Under these
+ * scenarios, we throw this special RuntimeException.
+ */
+public class BlockPushNonFatalFailure extends RuntimeException {
+
+ /**
+ * The error code of the failure, encoded as a ByteBuffer to be responded back to the client.
+ * Instead of responding a RPCFailure with the exception stack trace as the payload,
+ * which makes checking the content of the exception very tedious on the client side,
+ * we can respond a proper RPCResponse to make it more robust and efficient. This
+ * field is only set on the shuffle server side when the exception is originally generated.
+ */
+ private ByteBuffer response;
+
+ /**
+ * The error code of the failure. This field is only set on the client side when a
+ * BlockPushNonFatalFailure is recreated from the error code received from the server.
+ */
+ private ReturnCode returnCode;
+
+ public BlockPushNonFatalFailure(ByteBuffer response) {
+ this.response = response;
+ }
+
+ public BlockPushNonFatalFailure(ReturnCode returnCode) {
+ this.returnCode = returnCode;
+ }
+
+ /**
+ * Since this type of exception is used to only convey the error code, we reduce the
+ * exception initialization overhead by skipping filling the stack trace.
+ */
+ @Override
+ public synchronized Throwable fillInStackTrace() {
+ return this;
+ }
+
+ public ByteBuffer getResponse() {
+ return response;
+ }
+
+ public ReturnCode getReturnCode() {
+ return returnCode;
+ }
+
+ public enum ReturnCode {
+ /**
+ * Indicate the case of a successful merge of a pushed block.
+ */
+ SUCCESS(0),
Review comment:
Note: we are not gracefully handling this in ESS right now - but given we are doing a better job with this PR, wanted to move towards more robust code :-)
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] venkata91 commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
venkata91 commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r682308073
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -471,9 +488,10 @@ public void onData(String streamId, ByteBuffer buf) {
public void onComplete(String streamId) {
if (isStaleBlockOrTooLate) {
// Throw an exception here so the block data is drained from channel and server
- // responds RpcFailure to the client.
- throw new RuntimeException(String.format("Block %s %s", streamId,
- ErrorHandler.BlockPushErrorHandler.TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX));
+ // responds the error code to the client.
+ throw new BlockPushNonFatalFailure(
+ new PushBlockNonFatalErrorCode(ErrorCode.TOO_LATE_OR_STALE_BLOCK_PUSH.id())
Review comment:
I am not quite sure that we can separate out `too late` and `stale` pushes message separately from `RemoteBlockPushResolver`.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] otterc commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Minor changes to address a few identified push based shuffle server side inefficiencies.
Posted by GitBox <gi...@apache.org>.
otterc commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r681387486
##########
File path: core/src/main/scala/org/apache/spark/shuffle/ShuffleBlockPusher.scala
##########
@@ -78,10 +76,10 @@ private[spark] class ShuffleBlockPusher(conf: SparkConf) extends Logging {
if (t.getCause != null && t.getCause.isInstanceOf[FileNotFoundException]) {
return false
}
- val errorStackTraceString = Throwables.getStackTraceAsString(t)
// If the block is too late or the invalid block push, there is no need to retry it
- !errorStackTraceString.contains(
- BlockPushErrorHandler.TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX)
+ val msg = t.getMessage
+ !(msg != null && msg.contains(
+ BlockPushErrorHandler.TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX))
Review comment:
I think this should check the message in the cause? On the client side, the `RuntimeException` becomes the cause.
See TransportClient.uploadStream
```
RpcChannelListener listener = new RpcChannelListener(requestId, callback);
channel.writeAndFlush(new UploadStream(requestId, meta, data)).addListener(listener);
```
Here, RpcChannelListener wraps the exception with an IOException
```
@Override
void handleFailure(String errorMsg, Throwable cause) {
handler.removeRpcRequest(rpcRequestId);
callback.onFailure(new IOException(errorMsg, cause));
}
```
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] otterc commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Minor changes to address a few identified push based shuffle server side inefficiencies.
Posted by GitBox <gi...@apache.org>.
otterc commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r681392579
##########
File path: core/src/main/scala/org/apache/spark/shuffle/ShuffleBlockPusher.scala
##########
@@ -78,10 +76,10 @@ private[spark] class ShuffleBlockPusher(conf: SparkConf) extends Logging {
if (t.getCause != null && t.getCause.isInstanceOf[FileNotFoundException]) {
return false
}
- val errorStackTraceString = Throwables.getStackTraceAsString(t)
// If the block is too late or the invalid block push, there is no need to retry it
- !errorStackTraceString.contains(
- BlockPushErrorHandler.TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX)
+ val msg = t.getMessage
+ !(msg != null && msg.contains(
+ BlockPushErrorHandler.TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX))
Review comment:
Never mind, the `callback` that is added to `TranportResponseHandler` is the `BlockPushCallback`, so this is fine.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Victsm commented on pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
Victsm commented on pull request #33613:
URL: https://github.com/apache/spark/pull/33613#issuecomment-897045610
Looking at the test failure and the test itself, it seems that a `io.netty.channel.StacklessClosedChannelException` was thrown which is not in the hard-coded list of channel closed exceptions in `RPCIntegrationSuite`.
https://github.com/apache/spark/blob/ccead315b30764d10d8356fb0d5fce459f6e4f93/common/network-common/src/test/java/org/apache/spark/network/RpcIntegrationSuite.java#L383-L388
The fix could be to just add this exception type to the hardcoded list.
However, I also couldn't reproduce this test failure locally.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] venkata91 commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
venkata91 commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r682308073
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -471,9 +488,10 @@ public void onData(String streamId, ByteBuffer buf) {
public void onComplete(String streamId) {
if (isStaleBlockOrTooLate) {
// Throw an exception here so the block data is drained from channel and server
- // responds RpcFailure to the client.
- throw new RuntimeException(String.format("Block %s %s", streamId,
- ErrorHandler.BlockPushErrorHandler.TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX));
+ // responds the error code to the client.
+ throw new BlockPushNonFatalFailure(
+ new PushBlockNonFatalErrorCode(ErrorCode.TOO_LATE_OR_STALE_BLOCK_PUSH.id())
Review comment:
I am not quite sure whether it is possible to separate out `too late` and `stale` pushes separately.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33613:
URL: https://github.com/apache/spark/pull/33613#issuecomment-896201782
Kubernetes integration test unable to build dist.
exiting with code: 1
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/46792/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r683213429
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -164,10 +182,7 @@ private AppShufflePartitionInfo getOrCreateAppShufflePartitionInfo(
// current incoming one
int latestShuffleMergeId = appShuffleMergePartitionsInfo.shuffleMergeId;
if (latestShuffleMergeId > shuffleMergeId) {
- throw new StaleBlockPushException(String.format("Rejecting shuffle blocks push request"
- + " for shuffle %s with shuffleMergeId %s for application %s_%s as a higher"
- + " shuffleMergeId %s request is already seen", shuffleId, shuffleMergeId,
- appShuffleInfo.appId, appShuffleInfo.attemptId, latestShuffleMergeId));
Review comment:
log the error message?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Victsm commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
Victsm commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r682280865
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/OneForOneBlockPusher.java
##########
@@ -77,42 +80,55 @@ public OneForOneBlockPusher(
@Override
public void onSuccess(ByteBuffer response) {
- // On receipt of a successful block push
- listener.onBlockPushSuccess(blockId, new NioManagedBuffer(ByteBuffer.allocate(0)));
+ // If the response is non-empty, the server has responded some error code. Decode
+ // it and handle the error accordingly.
+ if (response.hasRemaining()) {
Review comment:
You are right that we would anyway incur the overhead with decoding an `RpcResponse` with every success response even if the body is empty.
Agree that having a dedicated code for success push makes it more consistent.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] otterc commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Minor changes to address a few identified push based shuffle server side inefficiencies.
Posted by GitBox <gi...@apache.org>.
otterc commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r681395236
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -863,8 +884,8 @@ public void onComplete(String streamId) throws IOException {
if (isStaleOrTooLate(appShuffleInfo.shuffles.get(partitionInfo.shuffleId),
partitionInfo.shuffleMergeId, partitionInfo.reduceId)) {
deferredBufs = null;
- throw new RuntimeException(String.format("Block %s is %s", streamId,
- ErrorHandler.BlockPushErrorHandler.TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX));
+ throw new RuntimeException("Block " + streamId + " " +
Review comment:
There are still other instances in this class where we are formatting the message of the RuntimeException with `String.format` but those haven't been modified. Is it because they are quite infrequent? Still would be nice to have this consistent in this class.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] otterc commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
otterc commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r682994361
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -471,9 +488,10 @@ public void onData(String streamId, ByteBuffer buf) {
public void onComplete(String streamId) {
if (isStaleBlockOrTooLate) {
// Throw an exception here so the block data is drained from channel and server
- // responds RpcFailure to the client.
- throw new RuntimeException(String.format("Block %s %s", streamId,
- ErrorHandler.BlockPushErrorHandler.TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX));
+ // responds the error code to the client.
+ throw new BlockPushNonFatalFailure(
+ new PushBlockNonFatalErrorCode(ErrorCode.TOO_LATE_OR_STALE_BLOCK_PUSH.id())
Review comment:
> For stale push from prior app attempt, it currently indeed throws an IllegalArgumentException inside receiveBlockDataAsStream which would lead to server closing channels.
I see. That's right. I missed that stale push here is used only in reference to the executors which are for the same app attempts but older merged id.
But anyways, I still think we should replace `StaleBlockPushException` usage with `BlockPushNonFatalFailure`.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Victsm commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
Victsm commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r682284968
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -471,9 +488,10 @@ public void onData(String streamId, ByteBuffer buf) {
public void onComplete(String streamId) {
if (isStaleBlockOrTooLate) {
// Throw an exception here so the block data is drained from channel and server
- // responds RpcFailure to the client.
- throw new RuntimeException(String.format("Block %s %s", streamId,
- ErrorHandler.BlockPushErrorHandler.TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX));
+ // responds the error code to the client.
+ throw new BlockPushNonFatalFailure(
+ new PushBlockNonFatalErrorCode(ErrorCode.TOO_LATE_OR_STALE_BLOCK_PUSH.id())
Review comment:
This is doable with some minor changes on the `RemoteBlockPushResolver` side to tell apart the two errors.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33613:
URL: https://github.com/apache/spark/pull/33613#issuecomment-895508258
Kubernetes integration test unable to build dist.
exiting with code: 1
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/46748/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] zhuqi-lucas commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
zhuqi-lucas commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r682301682
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -471,9 +488,10 @@ public void onData(String streamId, ByteBuffer buf) {
public void onComplete(String streamId) {
if (isStaleBlockOrTooLate) {
// Throw an exception here so the block data is drained from channel and server
- // responds RpcFailure to the client.
- throw new RuntimeException(String.format("Block %s %s", streamId,
- ErrorHandler.BlockPushErrorHandler.TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX));
+ // responds the error code to the client.
+ throw new BlockPushNonFatalFailure(
+ new PushBlockNonFatalErrorCode(ErrorCode.TOO_LATE_OR_STALE_BLOCK_PUSH.id())
Review comment:
I also think we'd better use error codes for "too late push" and "stale push" separately, and after this, i will fill the "too old attempt" case in [SPARK-35548](https://issues.apache.org/jira/browse/SPARK-35548).
##########
File path: pusher.diff
##########
@@ -0,0 +1,351 @@
+diff --git a/core/src/main/scala/org/apache/spark/shuffle/ShuffleBlockPusher.scala b/core/src/main/scala/org/apache/spark/shuffle/ShuffleBlockPusher.scala
Review comment:
@Victsm Is this file pushed by mistake?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Victsm commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
Victsm commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r686216518
##########
File path: common/network-common/src/main/java/org/apache/spark/network/server/BlockPushNonFatalFailure.java
##########
@@ -0,0 +1,162 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.network.server;
+
+import java.nio.ByteBuffer;
+
+import com.google.common.base.Preconditions;
+
+/**
+ * A special RuntimeException thrown when shuffle service experiences a non-fatal failure
+ * with handling block push requests with push-based shuffle. Due to the best-effort nature
+ * of push-based shuffle, there are cases where the exceptions gets thrown under certain
+ * relatively common cases such as when a pushed block is received after the corresponding
+ * shuffle is merge finalized or when a pushed block experiences merge collision. Under these
+ * scenarios, we throw this special RuntimeException.
+ */
+public class BlockPushNonFatalFailure extends RuntimeException {
+ /**
+ * String constant used for generating exception messages indicating a block to be merged
+ * arrives too late on the server side. When we get a block push failure because of the
+ * block arrives too late, we will not retry pushing the block nor log the exception on
+ * the client side.
+ */
+ public static final String TOO_LATE_BLOCK_PUSH_MESSAGE_SUFFIX =
+ " is received after merged shuffle is finalized";
+
+ /**
+ * String constant used for generating exception messages indicating a block to be merged
+ * is a stale block push in the case of indeterminate stage retries on the server side.
+ * When we get a block push failure because of the block push being stale, we will not
+ * retry pushing the block nor log the exception on the client side.
+ */
+ public static final String STALE_BLOCK_PUSH_MESSAGE_SUFFIX =
+ " is a stale block push from an indeterminate stage retry";
+
+ /**
+ * String constant used for generating exception messages indicating the server couldn't
+ * append a block after all available attempts due to collision with other blocks belonging
+ * to the same shuffle partition. When we get a block push failure because of the block
+ * couldn't be written due to this reason, we will not log the exception on the client side.
+ */
+ public static final String BLOCK_APPEND_COLLISION_MSG_SUFFIX =
+ " experienced merge collision on the server side";
+
+ /**
+ * The error code of the failure, encoded as a ByteBuffer to be responded back to the client.
+ * Instead of responding a RPCFailure with the exception stack trace as the payload,
+ * which makes checking the content of the exception very tedious on the client side,
+ * we can respond a proper RPCResponse to make it more robust and efficient. This
+ * field is only set on the shuffle server side when the exception is originally generated.
+ */
+ private ByteBuffer response;
+
+ /**
+ * The error code of the failure. This field is only set on the client side when a
+ * BlockPushNonFatalFailure is recreated from the error code received from the server.
+ */
+ private ReturnCode returnCode;
+
+ public BlockPushNonFatalFailure(ByteBuffer response, String msg) {
+ super(msg);
+ this.response = response;
+ }
+
+ public BlockPushNonFatalFailure(ReturnCode returnCode, String msg) {
+ super(msg);
+ this.returnCode = returnCode;
+ }
+
+ /**
+ * Since this type of exception is used to only convey the error code, we reduce the
+ * exception initialization overhead by skipping filling the stack trace.
+ */
+ @Override
+ public synchronized Throwable fillInStackTrace() {
+ return this;
+ }
+
+ public ByteBuffer getResponse() {
+ // Ensure we do not invoke this method if response is not set
+ Preconditions.checkNotNull(response);
+ return response;
+ }
+
+ public ReturnCode getReturnCode() {
+ // Ensure we do not invoke this method if returnCode is not set
+ Preconditions.checkNotNull(returnCode);
+ return returnCode;
+ }
+
+ public enum ReturnCode {
+ /**
+ * Indicate the case of a successful merge of a pushed block.
+ */
+ SUCCESS(0),
+ /**
+ * Indicate a block to be merged arrives too late on the server side, i.e. after the
+ * corresponding shuffle has been merge finalized. When the client gets this code, it
+ * will not retry pushing the block.
+ */
+ TOO_LATE_BLOCK_PUSH(1),
+ /**
+ * Indicating the server couldn't append a block after all available attempts due to
+ * collision with other blocks belonging to the same shuffle partition.
+ */
+ BLOCK_APPEND_COLLISION_DETECTED(2),
+ /**
+ * Indicate a block received on the server side is a stale block push in the case of
+ * indeterminate stage retries. When the client receives this code, it will not retry
+ * pushing the block.
+ */
+ STALE_BLOCK_PUSH(3);
+
+ private final byte id;
+
+ ReturnCode(int id) {
+ assert id < 128 : "Cannot have more than 128 block push return code";
+ this.id = (byte) id;
+ }
+
+ public byte id() { return id; }
+ }
+
+ public static ReturnCode getReturnCode(byte id) {
+ switch (id) {
+ case 0: return ReturnCode.SUCCESS;
+ case 1: return ReturnCode.TOO_LATE_BLOCK_PUSH;
+ case 2: return ReturnCode.BLOCK_APPEND_COLLISION_DETECTED;
+ case 3: return ReturnCode.STALE_BLOCK_PUSH;
+ default: throw new IllegalArgumentException("Unknown block push return code: " + id);
+ }
+ }
+
+ public static String getErrorMsg(String blockId, ReturnCode errorCode) {
+ Preconditions.checkArgument(errorCode != ReturnCode.SUCCESS);
+ switch (errorCode) {
+ case STALE_BLOCK_PUSH:
+ return "Block " + blockId + STALE_BLOCK_PUSH_MESSAGE_SUFFIX;
+ case TOO_LATE_BLOCK_PUSH:
+ return "Block " + blockId + TOO_LATE_BLOCK_PUSH_MESSAGE_SUFFIX;
+ case BLOCK_APPEND_COLLISION_DETECTED:
+ return "Block " + blockId + BLOCK_APPEND_COLLISION_MSG_SUFFIX;
Review comment:
Thanks for the suggestion. Modified the PR accordingly.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] otterc commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Minor changes to address a few identified push based shuffle server side inefficiencies.
Posted by GitBox <gi...@apache.org>.
otterc commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r681387486
##########
File path: core/src/main/scala/org/apache/spark/shuffle/ShuffleBlockPusher.scala
##########
@@ -78,10 +76,10 @@ private[spark] class ShuffleBlockPusher(conf: SparkConf) extends Logging {
if (t.getCause != null && t.getCause.isInstanceOf[FileNotFoundException]) {
return false
}
- val errorStackTraceString = Throwables.getStackTraceAsString(t)
// If the block is too late or the invalid block push, there is no need to retry it
- !errorStackTraceString.contains(
- BlockPushErrorHandler.TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX)
+ val msg = t.getMessage
+ !(msg != null && msg.contains(
+ BlockPushErrorHandler.TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX))
Review comment:
Should this not check the message in the cause? On the client side, the `RuntimeException` becomes the cause.
See TransportClient.uploadStream
```
RpcChannelListener listener = new RpcChannelListener(requestId, callback);
channel.writeAndFlush(new UploadStream(requestId, meta, data)).addListener(listener);
```
Here, RpcChannelListener wraps the exception with an IOException
```
@Override
void handleFailure(String errorMsg, Throwable cause) {
handler.removeRpcRequest(rpcRequestId);
callback.onFailure(new IOException(errorMsg, cause));
}
```
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on pull request #33613: [SPARK-36378][SHUFFLE] Minor changes to address a few identified push based shuffle server side inefficiencies.
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on pull request #33613:
URL: https://github.com/apache/spark/pull/33613#issuecomment-891622550
> 3. On the server side, the usage of Throwables.getStackTraceAsString in the ErrorHandler.shouldRetryError and ErrorHandler.shouldLogError has generated quite some overhead.
Should be client side?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] venkata91 commented on pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
venkata91 commented on pull request #33613:
URL: https://github.com/apache/spark/pull/33613#issuecomment-897010907
> @Victsm, @venkata91, @zhouyejoe Please take a look at @HyukjinKwon's comment above.
Ran `RpcIntegrationSuite` locally and it seems to succeed.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #33613:
URL: https://github.com/apache/spark/pull/33613#issuecomment-894706364
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/142191/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r683208602
##########
File path: common/network-common/src/main/java/org/apache/spark/network/server/BlockPushNonFatalFailure.java
##########
@@ -0,0 +1,114 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.network.server;
+
+import java.nio.ByteBuffer;
+
+/**
+ * A special RuntimeException thrown when shuffle service experiences a non-fatal failure
+ * with handling block push requests with push-based shuffle. Due to the best-effort nature
+ * of push-based shuffle, there are cases where the exceptions gets thrown under certain
+ * relatively common cases such as when a pushed block is received after the corresponding
+ * shuffle is merge finalized or when a pushed block experiences merge collision. Under these
+ * scenarios, we throw this special RuntimeException.
+ */
+public class BlockPushNonFatalFailure extends RuntimeException {
+
+ /**
+ * The error code of the failure, encoded as a ByteBuffer to be responded back to the client.
+ * Instead of responding a RPCFailure with the exception stack trace as the payload,
+ * which makes checking the content of the exception very tedious on the client side,
+ * we can respond a proper RPCResponse to make it more robust and efficient. This
+ * field is only set on the shuffle server side when the exception is originally generated.
+ */
+ private ByteBuffer response;
+
+ /**
+ * The error code of the failure. This field is only set on the client side when a
+ * BlockPushNonFatalFailure is recreated from the error code received from the server.
+ */
+ private ReturnCode returnCode;
+
+ public BlockPushNonFatalFailure(ByteBuffer response) {
+ this.response = response;
+ }
+
+ public BlockPushNonFatalFailure(ReturnCode returnCode) {
+ this.returnCode = returnCode;
+ }
+
+ /**
+ * Since this type of exception is used to only convey the error code, we reduce the
+ * exception initialization overhead by skipping filling the stack trace.
+ */
+ @Override
+ public synchronized Throwable fillInStackTrace() {
+ return this;
+ }
+
+ public ByteBuffer getResponse() {
+ return response;
+ }
+
+ public ReturnCode getReturnCode() {
+ return returnCode;
+ }
+
+ public enum ReturnCode {
+ /**
+ * Indicate the case of a successful merge of a pushed block.
+ */
+ SUCCESS(0),
Review comment:
`UNKNOWN` sgtm.
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -164,10 +182,7 @@ private AppShufflePartitionInfo getOrCreateAppShufflePartitionInfo(
// current incoming one
int latestShuffleMergeId = appShuffleMergePartitionsInfo.shuffleMergeId;
if (latestShuffleMergeId > shuffleMergeId) {
- throw new StaleBlockPushException(String.format("Rejecting shuffle blocks push request"
- + " for shuffle %s with shuffleMergeId %s for application %s_%s as a higher"
- + " shuffleMergeId %s request is already seen", shuffleId, shuffleMergeId,
- appShuffleInfo.appId, appShuffleInfo.attemptId, latestShuffleMergeId));
Review comment:
log the error message?
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -164,10 +182,7 @@ private AppShufflePartitionInfo getOrCreateAppShufflePartitionInfo(
// current incoming one
int latestShuffleMergeId = appShuffleMergePartitionsInfo.shuffleMergeId;
if (latestShuffleMergeId > shuffleMergeId) {
- throw new StaleBlockPushException(String.format("Rejecting shuffle blocks push request"
- + " for shuffle %s with shuffleMergeId %s for application %s_%s as a higher"
- + " shuffleMergeId %s request is already seen", shuffleId, shuffleMergeId,
- appShuffleInfo.appId, appShuffleInfo.attemptId, latestShuffleMergeId));
Review comment:
Could we also log for the case of:
```
if (dataFile.exists()) {
return null;
}
```
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -164,10 +182,7 @@ private AppShufflePartitionInfo getOrCreateAppShufflePartitionInfo(
// current incoming one
int latestShuffleMergeId = appShuffleMergePartitionsInfo.shuffleMergeId;
if (latestShuffleMergeId > shuffleMergeId) {
- throw new StaleBlockPushException(String.format("Rejecting shuffle blocks push request"
- + " for shuffle %s with shuffleMergeId %s for application %s_%s as a higher"
- + " shuffleMergeId %s request is already seen", shuffleId, shuffleMergeId,
- appShuffleInfo.appId, appShuffleInfo.attemptId, latestShuffleMergeId));
Review comment:
Could we also log for the case of:
```
if (dataFile.exists()) {
return null;
}
```
Btw, is it equivalent to throw `BlockPushNonFatalFailure(ReturnCode.TOO_LATE_BLOCK_PUSH)` rather than return null? If it's equivalent, I think throwing exception is more consistent with the case of "stale push"
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -385,14 +397,24 @@ public StreamCallbackWithID receiveBlockDataAsStream(PushBlockStream msg) {
+ "with the current attempt id %s stored in shuffle service for application %s",
msg.appAttemptId, appShuffleInfo.attemptId, msg.appId));
}
+ // Use string concatenation here to avoid the overhead with String.format on every
+ // pushed block.
+ final String streamId = OneForOneBlockPusher.SHUFFLE_PUSH_BLOCK_PREFIX + "_"
+ + msg.shuffleId + "_" + msg.shuffleMergeId + "_" + msg.mapIndex + "_" + msg.reduceId;
// Retrieve merged shuffle file metadata
AppShufflePartitionInfo partitionInfoBeforeCheck;
+ boolean isStaleBlock = false;
+ boolean isTooLate = false;
Review comment:
Why not just use the `ReturnCode` filed directly?
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/OneForOneBlockPusher.java
##########
@@ -77,42 +81,55 @@ public OneForOneBlockPusher(
@Override
public void onSuccess(ByteBuffer response) {
- // On receipt of a successful block push
- listener.onBlockPushSuccess(blockId, new NioManagedBuffer(ByteBuffer.allocate(0)));
+ BlockPushReturnCode returnCode =
+ (BlockPushReturnCode) BlockTransferMessage.Decoder.fromByteBuffer(response);
+ // If the return code is not SUCCESS, the server has responded some error code. Handle
+ // the error accordingly.
+ if (returnCode.returnCode != ReturnCode.SUCCESS.id()) {
+ checkAndFailRemainingBlocks(index, new BlockPushNonFatalFailure(
+ BlockPushNonFatalFailure.getReturnCode(returnCode.returnCode)));
+ } else {
+ // On receipt of a successful block push
+ listener.onBlockPushSuccess(blockId, new NioManagedBuffer(ByteBuffer.allocate(0)));
+ }
}
@Override
public void onFailure(Throwable e) {
- // Since block push is best effort, i.e., if we encounter a block push failure that's still
- // retriable according to ErrorHandler (not a connection exception and the block is not too
- // late), we should not fail all remaining block pushes even though
- // RetryingBlockTransferor might consider this failure not retriable (exceeding max retry
- // count etc). The best effort nature makes block push tolerable of a partial completion.
- // Thus, we only fail the block that's actually failed in this case. Note that, on the
- // RetryingBlockTransferor side, if retry is initiated, it would still invalidate the
- // previous active retry listener, and retry pushing all outstanding blocks. However, since
- // the blocks to be pushed are preloaded into memory and the first attempt of pushing these
- // blocks might have already succeeded, retry pushing all the outstanding blocks should be
- // very cheap (on the client side, the block data is in memory; on the server side, the block
- // will be recognized as a duplicate which triggers noop handling). Here, by failing only the
- // one block that's actually failed, we are essentially preventing forwarding unnecessary
- // block push failures to the parent listener of the retry listener.
- //
- // Take the following as an example. For the common exception during block push handling,
- // i.e. block collision, it is considered as retriable by ErrorHandler but not retriable
- // by RetryingBlockTransferor. When we encounter a failure of this type, we only fail the
- // one block encountering this issue not the remaining blocks in the same batch. On the
- // RetryingBlockTransferor side, since this exception is considered as not retriable, it
- // would immediately invoke parent listener's onBlockTransferFailure. However, the remaining
- // blocks in the same batch would remain current and active and they won't be impacted by
- // this exception.
- if (PUSH_ERROR_HANDLER.shouldRetryError(e)) {
- String[] targetBlockId = Arrays.copyOfRange(blockIds, index, index + 1);
- failRemainingBlocks(targetBlockId, e);
- } else {
- String[] targetBlockId = Arrays.copyOfRange(blockIds, index, blockIds.length);
- failRemainingBlocks(targetBlockId, e);
- }
+ checkAndFailRemainingBlocks(index, e);
+ }
+ }
+
+ private void checkAndFailRemainingBlocks(int index, Throwable e) {
+ // Since block push is best effort, i.e., if we encounter a block push failure that's still
+ // retriable according to ErrorHandler (not a connection exception and the block is not too
+ // late), we should not fail all remaining block pushes even though
+ // RetryingBlockTransferor might consider this failure not retriable (exceeding max retry
+ // count etc). The best effort nature makes block push tolerable of a partial completion.
+ // Thus, we only fail the block that's actually failed in this case. Note that, on the
+ // RetryingBlockTransferor side, if retry is initiated, it would still invalidate the
+ // previous active retry listener, and retry pushing all outstanding blocks. However, since
+ // the blocks to be pushed are preloaded into memory and the first attempt of pushing these
+ // blocks might have already succeeded, retry pushing all the outstanding blocks should be
+ // very cheap (on the client side, the block data is in memory; on the server side, the block
+ // will be recognized as a duplicate which triggers noop handling). Here, by failing only the
+ // one block that's actually failed, we are essentially preventing forwarding unnecessary
+ // block push failures to the parent listener of the retry listener.
+ //
+ // Take the following as an example. For the common exception during block push handling,
+ // i.e. block collision, it is considered as retriable by ErrorHandler but not retriable
+ // by RetryingBlockTransferor. When we encounter a failure of this type, we only fail the
+ // one block encountering this issue not the remaining blocks in the same batch. On the
+ // RetryingBlockTransferor side, since this exception is considered as not retriable, it
+ // would immediately invoke parent listener's onBlockTransferFailure. However, the remaining
+ // blocks in the same batch would remain current and active and they won't be impacted by
+ // this exception.
Review comment:
Shall we log the specific error for each case? Also, including whether we'd retry the block or not.
##########
File path: common/network-common/src/main/java/org/apache/spark/network/server/TransportRequestHandler.java
##########
@@ -213,7 +213,15 @@ public void onData(String streamId, ByteBuffer buf) throws IOException {
public void onComplete(String streamId) throws IOException {
try {
streamHandler.onComplete(streamId);
- callback.onSuccess(ByteBuffer.allocate(0));
+ callback.onSuccess(streamHandler.getCompletionResponse());
+ } catch (BlockPushNonFatalFailure ex) {
Review comment:
Shall we catch `BlockPushNonFatalFailure` inside the `onComplete`? We can have a response filed inside `streamHandler` to store the response value. In this way, `getCompletionResponse` can handle the success and error cases consistently.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #33613:
URL: https://github.com/apache/spark/pull/33613#issuecomment-894694385
**[Test build #142191 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/142191/testReport)** for PR 33613 at commit [`51841e4`](https://github.com/apache/spark/commit/51841e45972086aa08a2c23f227a779bf8e8daf0).
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Victsm commented on pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
Victsm commented on pull request #33613:
URL: https://github.com/apache/spark/pull/33613#issuecomment-897055003
Checked on the Netty side.
So this `io.netty.channel.StacklessClosedChannelException` was introduced starting from Netty 4.1.52.
In April, Spark's Netty dependency got bumped from 4.1.51 to 4.1.63.
So I believe this issue started happening intermittently since that dependency version bump, and we indeed need to add this new exception type to the hardcoded list.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
mridulm commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r683089989
##########
File path: common/network-common/src/main/java/org/apache/spark/network/server/BlockPushNonFatalFailure.java
##########
@@ -0,0 +1,114 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.network.server;
+
+import java.nio.ByteBuffer;
+
+/**
+ * A special RuntimeException thrown when shuffle service experiences a non-fatal failure
+ * with handling block push requests with push-based shuffle. Due to the best-effort nature
+ * of push-based shuffle, there are cases where the exceptions gets thrown under certain
+ * relatively common cases such as when a pushed block is received after the corresponding
+ * shuffle is merge finalized or when a pushed block experiences merge collision. Under these
+ * scenarios, we throw this special RuntimeException.
+ */
+public class BlockPushNonFatalFailure extends RuntimeException {
+
+ /**
+ * The error code of the failure, encoded as a ByteBuffer to be responded back to the client.
+ * Instead of responding a RPCFailure with the exception stack trace as the payload,
+ * which makes checking the content of the exception very tedious on the client side,
+ * we can respond a proper RPCResponse to make it more robust and efficient. This
+ * field is only set on the shuffle server side when the exception is originally generated.
+ */
+ private ByteBuffer response;
+
+ /**
+ * The error code of the failure. This field is only set on the client side when a
+ * BlockPushNonFatalFailure is recreated from the error code received from the server.
+ */
+ private ReturnCode returnCode;
+
+ public BlockPushNonFatalFailure(ByteBuffer response) {
+ this.response = response;
+ }
+
+ public BlockPushNonFatalFailure(ReturnCode returnCode) {
+ this.returnCode = returnCode;
+ }
+
+ /**
+ * Since this type of exception is used to only convey the error code, we reduce the
+ * exception initialization overhead by skipping filling the stack trace.
+ */
+ @Override
+ public synchronized Throwable fillInStackTrace() {
+ return this;
+ }
+
+ public ByteBuffer getResponse() {
Review comment:
Add `Preconditions.checkNotNull` here and in `getReturnCode` to ensure incorrect usage is flagged.
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalBlockStoreClient.java
##########
@@ -218,7 +218,7 @@ public void getMergedBlockMeta(
public void onSuccess(int numChunks, ManagedBuffer buffer) {
logger.trace("Successfully got merged block meta for shuffleId {} shuffleMergeId {}"
+ " reduceId {}", shuffleId, shuffleMergeId, reduceId);
- listener.onSuccess(shuffleId, reduceId, shuffleMergeId,
+ listener.onSuccess(shuffleId, shuffleMergeId, reduceId,
Review comment:
Nice catch !
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -669,6 +696,11 @@ public String getID() {
return streamId;
}
+ @Override
+ public ByteBuffer getCompletionResponse() {
+ return new BlockPushReturnCode(ReturnCode.SUCCESS.id()).toByteBuffer();
+ }
+
Review comment:
nit:
Given how frequently this will be called - make this into a static final read only bytebuffer we duplicate each time it is required ?
Something like:
```
private static final ByteBuffer SUCCESS_BUFFER =
new BlockPushReturnCode(ReturnCode.SUCCESS.id()).toByteBuffer().asReadOnlyBuffer();
```
and getCompletionResponse -> `SUCCESS_BUFFER.duplicate()`
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -373,9 +386,10 @@ void deleteExecutorDirs(AppShuffleInfo appShuffleInfo) {
@Override
public StreamCallbackWithID receiveBlockDataAsStream(PushBlockStream msg) {
AppShuffleInfo appShuffleInfo = validateAndGetAppShuffleInfo(msg.appId);
- final String streamId = String.format("%s_%d_%d_%d_%d",
- OneForOneBlockPusher.SHUFFLE_PUSH_BLOCK_PREFIX, msg.shuffleId, msg.shuffleMergeId,
- msg.mapIndex, msg.reduceId);
+ // Use string concatenation here to avoid the overhead with String.format on every
+ // pushed block.
+ final String streamId = OneForOneBlockPusher.SHUFFLE_PUSH_BLOCK_PREFIX + "_"
+ + msg.shuffleId + "_" + msg.shuffleMergeId + "_" + msg.mapIndex + "_" + msg.reduceId;
if (appShuffleInfo.attemptId != msg.appAttemptId) {
Review comment:
Move this to after the attemptId check.
##########
File path: common/network-common/src/main/java/org/apache/spark/network/server/BlockPushNonFatalFailure.java
##########
@@ -0,0 +1,114 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.network.server;
+
+import java.nio.ByteBuffer;
+
+/**
+ * A special RuntimeException thrown when shuffle service experiences a non-fatal failure
+ * with handling block push requests with push-based shuffle. Due to the best-effort nature
+ * of push-based shuffle, there are cases where the exceptions gets thrown under certain
+ * relatively common cases such as when a pushed block is received after the corresponding
+ * shuffle is merge finalized or when a pushed block experiences merge collision. Under these
+ * scenarios, we throw this special RuntimeException.
+ */
+public class BlockPushNonFatalFailure extends RuntimeException {
+
+ /**
+ * The error code of the failure, encoded as a ByteBuffer to be responded back to the client.
+ * Instead of responding a RPCFailure with the exception stack trace as the payload,
+ * which makes checking the content of the exception very tedious on the client side,
+ * we can respond a proper RPCResponse to make it more robust and efficient. This
+ * field is only set on the shuffle server side when the exception is originally generated.
+ */
+ private ByteBuffer response;
+
+ /**
+ * The error code of the failure. This field is only set on the client side when a
+ * BlockPushNonFatalFailure is recreated from the error code received from the server.
+ */
+ private ReturnCode returnCode;
+
+ public BlockPushNonFatalFailure(ByteBuffer response) {
+ this.response = response;
+ }
+
+ public BlockPushNonFatalFailure(ReturnCode returnCode) {
+ this.returnCode = returnCode;
+ }
+
+ /**
+ * Since this type of exception is used to only convey the error code, we reduce the
+ * exception initialization overhead by skipping filling the stack trace.
+ */
+ @Override
+ public synchronized Throwable fillInStackTrace() {
+ return this;
+ }
+
+ public ByteBuffer getResponse() {
+ return response;
+ }
+
+ public ReturnCode getReturnCode() {
+ return returnCode;
+ }
+
+ public enum ReturnCode {
+ /**
+ * Indicate the case of a successful merge of a pushed block.
+ */
+ SUCCESS(0),
Review comment:
Add a marker at index `0` for `UNKNOWN(0)` or `INVALID(0)` - to help with potential incompatible evolutions.
Thoughts @Ngone51 ?
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -119,7 +122,20 @@ public RemoteBlockPushResolver(TransportConf conf) {
.maximumWeight(conf.mergedIndexCacheSize())
.weigher((Weigher<File, ShuffleIndexInformation>)(file, indexInfo) -> indexInfo.getSize())
.build(ShuffleIndexInformation::new);
- this.errorHandler = new ErrorHandler.BlockPushErrorHandler();
+ this.errorHandler = createErrorHandler();
+ }
+
+ @VisibleForTesting
+ protected ErrorHandler.BlockPushErrorHandler createErrorHandler() {
+ return new ErrorHandler.BlockPushErrorHandler() {
+ // Explicitly use a shuffle service side error handler for handling exceptions.
+ // BlockPushNonException on the server side only has the response field set. It
+ // might require different handling logic compared with a client side error handler.
+ @Override
+ public boolean shouldLogError(Throwable t) {
+ return !(t instanceof BlockPushNonFatalFailure);
+ }
+ };
Review comment:
nit: Make this a static final field and return same instance always
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -379,9 +399,10 @@ void deleteExecutorDirs(AppShuffleInfo appShuffleInfo) {
@Override
public StreamCallbackWithID receiveBlockDataAsStream(PushBlockStream msg) {
AppShuffleInfo appShuffleInfo = validateAndGetAppShuffleInfo(msg.appId);
- final String streamId = String.format("%s_%d_%d_%d_%d",
- OneForOneBlockPusher.SHUFFLE_PUSH_BLOCK_PREFIX, msg.shuffleId, msg.shuffleMergeId,
- msg.mapIndex, msg.reduceId);
+ // Use string concatenation here to avoid the overhead with String.format on every
Review comment:
+1 to avoid `String.format` for simple patterns (unless it is being lazily evaluated on demand).
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/OneForOneBlockPusher.java
##########
@@ -77,42 +80,55 @@ public OneForOneBlockPusher(
@Override
public void onSuccess(ByteBuffer response) {
- // On receipt of a successful block push
- listener.onBlockPushSuccess(blockId, new NioManagedBuffer(ByteBuffer.allocate(0)));
+ BlockPushReturnCode returnCode =
+ (BlockPushReturnCode) BlockTransferMessage.Decoder.fromByteBuffer(response);
+ // If the return code is not 0, the server has responded some error code. Handle
+ // the error accordingly.
+ if (returnCode.returnCode > 0) {
Review comment:
`ReturnCode.SUCCESS != returnCode.returnCode`
##########
File path: core/src/main/scala/org/apache/spark/shuffle/ShuffleBlockPusher.scala
##########
@@ -242,10 +244,16 @@ private[spark] class ShuffleBlockPusher(conf: SparkConf) extends Logging {
handleResult(PushResult(blockId, exception))
}
}
+ // In addition to randomizing the order of the push requests, further randomize the order
+ // of blocks within the push request to further reduce the likelihood of shuffle server side
+ // collision of pushed blocks. This does not increase the cost of reading unmerged shuffle
+ // files on the executor side, because we are still reading MB-size chunks and only randomize
+ // the in-memory sliced buffers post reading.
+ val blocksToPush = Utils.randomize(blockIds.zip(
+ sliceReqBufferIntoBlockBuffers(request.reqBuffer, request.blocks.map(_._2))))
Review comment:
Move this to a different PR - this is not related to the current change.
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -781,8 +819,10 @@ public void onData(String streamId, ByteBuffer buf) throws IOException {
// to disk as well. This way, we avoid having to buffer the entirety of every blocks in
// memory, while still providing the necessary guarantee.
synchronized (partitionInfo) {
- if (isStaleOrTooLate(appShuffleInfo.shuffles.get(partitionInfo.shuffleId),
- partitionInfo.shuffleMergeId, partitionInfo.reduceId)) {
+ if (isStale(appShuffleInfo.shuffles.get(partitionInfo.shuffleId),
+ partitionInfo.shuffleMergeId) ||
+ isTooLate(appShuffleInfo.shuffles.get(partitionInfo.shuffleId),
+ partitionInfo.reduceId)) {
Review comment:
Fetch `appShuffleInfo.shuffles.get(partitionInfo.shuffleId)` and use same instance for both checks.
This pattern is there in `onComplete` and `onFailure` as well.
##########
File path: pusher.diff
##########
@@ -0,0 +1,351 @@
+diff --git a/core/src/main/scala/org/apache/spark/shuffle/ShuffleBlockPusher.scala b/core/src/main/scala/org/apache/spark/shuffle/ShuffleBlockPusher.scala
Review comment:
Looks like it @zhuqi-lucas - please remove this file @Victsm
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -436,17 +456,16 @@ public StreamCallbackWithID receiveBlockDataAsStream(PushBlockStream msg) {
// getting killed. When this happens, we need to distinguish the duplicate blocks as they
// arrive. More details on this is explained in later comments.
- // Track if the block is received after shuffle merge finalized or from an older
- // shuffleMergeId attempt.
- final boolean isStaleBlockOrTooLate = partitionInfoBeforeCheck == null;
// Check if the given block is already merged by checking the bitmap against the given map
// index
- final AppShufflePartitionInfo partitionInfo = isStaleBlockOrTooLate ? null :
+ final AppShufflePartitionInfo partitionInfo = isStaleBlock || isTooLate ? null :
partitionInfoBeforeCheck.mapTracker.contains(msg.mapIndex) ? null : partitionInfoBeforeCheck;
if (partitionInfo != null) {
return new PushBlockStreamCallback(
this, appShuffleInfo, streamId, partitionInfo, msg.mapIndex);
} else {
+ final boolean finalIsTooLate = isTooLate;
+ final boolean finalIsStaleBlock = isStaleBlock;
Review comment:
```suggestion
final BlockPushNonFatalFailure exception = isTooLate ? createTooLate .. : (isStaleBlock ? createStale .. : null);
```
and `if (null != exception) throw exception` in onComplete handler.
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/protocol/BlockPushReturnCode.java
##########
@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.network.shuffle.protocol;
+
+import java.util.Objects;
+
+import io.netty.buffer.ByteBuf;
+import org.apache.commons.lang3.builder.ToStringBuilder;
+import org.apache.commons.lang3.builder.ToStringStyle;
+
+/**
+ * Error code indicating a non-fatal failure of a block push request.
+ * Due to the best-effort nature of push-based shuffle, these failures
+ * do not impact the completion of the block push process. The list of
+ * such errors is in
+ * {@link org.apache.spark.network.server.BlockPushNonFatalFailure.ReturnCode}.
+ *
+ * @since 3.2.0
+ */
+public class BlockPushReturnCode extends BlockTransferMessage {
+ public final byte returnCode;
+
+ public BlockPushReturnCode(byte returnCode) {
+ this.returnCode = returnCode;
Review comment:
Validate if the returnCode is supported.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Minor changes to address a few identified push based shuffle server side inefficiencies.
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r681549698
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ErrorHandler.java
##########
@@ -105,16 +105,16 @@ public boolean shouldRetryError(Throwable t) {
return false;
}
- String errorStackTrace = Throwables.getStackTraceAsString(t);
// If the block is too late or stale block push, there is no need to retry it
- return !errorStackTrace.contains(TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX);
+ String msg = t.getMessage();
+ return !(msg != null && msg.contains(TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX));
Review comment:
I really think the way to matching the target string is kind of fragile. It potentially assumes the structure of the error. But I think we can not make sure that the error structure would never change.
So I'm thinking about using an enum type to represent different push results and reply with a normal RPC message.
WDYT?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] otterc commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
otterc commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r682876739
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -471,9 +488,10 @@ public void onData(String streamId, ByteBuffer buf) {
public void onComplete(String streamId) {
if (isStaleBlockOrTooLate) {
// Throw an exception here so the block data is drained from channel and server
- // responds RpcFailure to the client.
- throw new RuntimeException(String.format("Block %s %s", streamId,
- ErrorHandler.BlockPushErrorHandler.TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX));
+ // responds the error code to the client.
+ throw new BlockPushNonFatalFailure(
+ new PushBlockNonFatalErrorCode(ErrorCode.TOO_LATE_OR_STALE_BLOCK_PUSH.id())
Review comment:
I saw how it was being used. My point is that this is not needed any more. We can use `BlockPushNoneFatalFailure` instead of `StaleBlockPushException`.
Also, catching the `StaleBlockPushException` is not really required. I am pointing to the below lines. If `getOrCreateAppShufflePartitionInfo` throws a `RuntimeException`, then `receiveBlockDataAsStream` would just fail. In case of stale push, it's not really needed that a `StreamCallback` is returned.
```
try {
partitionInfoBeforeCheck = getOrCreateAppShufflePartitionInfo(appShuffleInfo, msg.shuffleId,
msg.shuffleMergeId, msg.reduceId);
} catch(StaleBlockPushException sbp) {
// Set partitionInfoBeforeCheck to null so that stale block push gets handled.
partitionInfoBeforeCheck = null;
}
```
So, this could just simplified with
```
AppShufflePartitionInfo partitionInfoBeforeCheck = getOrCreateAppShufflePartitionInfo(appShuffleInfo, msg.shuffleId,
msg.shuffleMergeId, msg.reduceId);
```
If the execution goes beyond this, then that means the push was not stale and then it is just too late.
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -471,9 +488,10 @@ public void onData(String streamId, ByteBuffer buf) {
public void onComplete(String streamId) {
if (isStaleBlockOrTooLate) {
// Throw an exception here so the block data is drained from channel and server
- // responds RpcFailure to the client.
- throw new RuntimeException(String.format("Block %s %s", streamId,
- ErrorHandler.BlockPushErrorHandler.TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX));
+ // responds the error code to the client.
+ throw new BlockPushNonFatalFailure(
+ new PushBlockNonFatalErrorCode(ErrorCode.TOO_LATE_OR_STALE_BLOCK_PUSH.id())
Review comment:
I saw how it was being used. My point is that this is not needed any more. We can use `BlockPushNoneFatalFailure` instead of `StaleBlockPushException` where it is instantiated.
Also, catching the `StaleBlockPushException` is not really required. I am pointing to the below lines. If `getOrCreateAppShufflePartitionInfo` throws a `RuntimeException`, then `receiveBlockDataAsStream` would just fail. In case of stale push, it's not really needed that a `StreamCallback` is returned.
```
try {
partitionInfoBeforeCheck = getOrCreateAppShufflePartitionInfo(appShuffleInfo, msg.shuffleId,
msg.shuffleMergeId, msg.reduceId);
} catch(StaleBlockPushException sbp) {
// Set partitionInfoBeforeCheck to null so that stale block push gets handled.
partitionInfoBeforeCheck = null;
}
```
So, this could just simplified with
```
AppShufflePartitionInfo partitionInfoBeforeCheck = getOrCreateAppShufflePartitionInfo(appShuffleInfo, msg.shuffleId,
msg.shuffleMergeId, msg.reduceId);
```
If the execution goes beyond this, then that means the push was not stale and then it is just too late.
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -471,9 +488,10 @@ public void onData(String streamId, ByteBuffer buf) {
public void onComplete(String streamId) {
if (isStaleBlockOrTooLate) {
// Throw an exception here so the block data is drained from channel and server
- // responds RpcFailure to the client.
- throw new RuntimeException(String.format("Block %s %s", streamId,
- ErrorHandler.BlockPushErrorHandler.TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX));
+ // responds the error code to the client.
+ throw new BlockPushNonFatalFailure(
+ new PushBlockNonFatalErrorCode(ErrorCode.TOO_LATE_OR_STALE_BLOCK_PUSH.id())
Review comment:
@Victsm that's a good point. But we can still replace the usage of `StaleBlockPushException` which is created with a message that uses `String.format` with `BlockPushNonFatalFailure`. We can still catch `BlockPushNonFatalFailure` and based on the `ErrorCode` enum differentiate between stale and too late.
Also, for stale block pushes that are for prior application attempts, isn't it closing the channel better? For clients that are pushing blocks from prior application attempt, we just want them to stop pushing all together correct.
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -471,9 +488,10 @@ public void onData(String streamId, ByteBuffer buf) {
public void onComplete(String streamId) {
if (isStaleBlockOrTooLate) {
// Throw an exception here so the block data is drained from channel and server
- // responds RpcFailure to the client.
- throw new RuntimeException(String.format("Block %s %s", streamId,
- ErrorHandler.BlockPushErrorHandler.TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX));
+ // responds the error code to the client.
+ throw new BlockPushNonFatalFailure(
+ new PushBlockNonFatalErrorCode(ErrorCode.TOO_LATE_OR_STALE_BLOCK_PUSH.id())
Review comment:
@Victsm that's a good point but we can still replace the usage of `StaleBlockPushException` which is created with a message that uses `String.format` with `BlockPushNonFatalFailure`. We can still catch `BlockPushNonFatalFailure` and based on the `ErrorCode` enum differentiate between stale and too late.
Also, for stale block pushes that are for prior application attempts, isn't it closing the channel better? For clients that are pushing blocks from prior application attempt, we just want them to stop pushing all together correct?
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -471,9 +488,10 @@ public void onData(String streamId, ByteBuffer buf) {
public void onComplete(String streamId) {
if (isStaleBlockOrTooLate) {
// Throw an exception here so the block data is drained from channel and server
- // responds RpcFailure to the client.
- throw new RuntimeException(String.format("Block %s %s", streamId,
- ErrorHandler.BlockPushErrorHandler.TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX));
+ // responds the error code to the client.
+ throw new BlockPushNonFatalFailure(
+ new PushBlockNonFatalErrorCode(ErrorCode.TOO_LATE_OR_STALE_BLOCK_PUSH.id())
Review comment:
@Victsm that's a good point but we can still replace the usage of `StaleBlockPushException` which is created with a message that uses `String.format` with `BlockPushNonFatalFailure`. We can still catch `BlockPushNonFatalFailure` and based on the `ErrorCode` enum differentiate between stale and too late.
Also, for stale block pushes that are for prior application attempts, isn't closing the channel better? For clients that are pushing blocks from prior application attempt, we just want them to stop pushing all together correct?
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -471,9 +488,10 @@ public void onData(String streamId, ByteBuffer buf) {
public void onComplete(String streamId) {
if (isStaleBlockOrTooLate) {
// Throw an exception here so the block data is drained from channel and server
- // responds RpcFailure to the client.
- throw new RuntimeException(String.format("Block %s %s", streamId,
- ErrorHandler.BlockPushErrorHandler.TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX));
+ // responds the error code to the client.
+ throw new BlockPushNonFatalFailure(
+ new PushBlockNonFatalErrorCode(ErrorCode.TOO_LATE_OR_STALE_BLOCK_PUSH.id())
Review comment:
> For stale push from prior app attempt, it currently indeed throws an IllegalArgumentException inside receiveBlockDataAsStream which would lead to server closing channels.
I see. That's right. I missed that stale push here is used only in reference to the executors which are for the same app attempts but older merged id.
But anyways, I still think we should replace `StaleBlockPushException` usage with `BlockPushNonFatalFailure`.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #33613:
URL: https://github.com/apache/spark/pull/33613#issuecomment-896287094
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/46797/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
mridulm commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r684290557
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/OneForOneBlockPusher.java
##########
@@ -77,42 +81,55 @@ public OneForOneBlockPusher(
@Override
public void onSuccess(ByteBuffer response) {
- // On receipt of a successful block push
- listener.onBlockPushSuccess(blockId, new NioManagedBuffer(ByteBuffer.allocate(0)));
+ BlockPushReturnCode returnCode =
+ (BlockPushReturnCode) BlockTransferMessage.Decoder.fromByteBuffer(response);
+ // If the return code is not SUCCESS, the server has responded some error code. Handle
+ // the error accordingly.
+ if (returnCode.returnCode != ReturnCode.SUCCESS.id()) {
+ checkAndFailRemainingBlocks(index, new BlockPushNonFatalFailure(
+ BlockPushNonFatalFailure.getReturnCode(returnCode.returnCode)));
Review comment:
Move this as a static helper method into `BlockPushNonFatalFailure` ? It is a repeating pattern to create `BlockPushNonFatalFailure` given a `ReturnCode` id.
Something like `BlockPushNonFatalFailure.fromReturnCode`
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] otterc commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Minor changes to address a few identified push based shuffle server side inefficiencies.
Posted by GitBox <gi...@apache.org>.
otterc commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r681927734
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ErrorHandler.java
##########
@@ -105,16 +105,16 @@ public boolean shouldRetryError(Throwable t) {
return false;
}
- String errorStackTrace = Throwables.getStackTraceAsString(t);
// If the block is too late or stale block push, there is no need to retry it
- return !errorStackTrace.contains(TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX);
+ String msg = t.getMessage();
+ return !(msg != null && msg.contains(TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX));
Review comment:
Had some discussions related to making this a proper RPC response with @Victsm in the past and I think as well that it is a better approach. One thing to note though is that if we do this, we will have to make changes on the client to invoke `listener.onFailure()` callback when the RPCResponse represents these exception types because on the client we leverage `RetryingBlockTransfer`/ `BlockTransferListener` to retry block push which relies on the `onFailure` callbacks.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #33613: [SPARK-36378][SHUFFLE] Minor changes to address a few identified push based shuffle server side inefficiencies.
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #33613:
URL: https://github.com/apache/spark/pull/33613#issuecomment-891405357
Can one of the admins verify this patch?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Victsm commented on pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
Victsm commented on pull request #33613:
URL: https://github.com/apache/spark/pull/33613#issuecomment-897063107
@HyukjinKwon @mridulm , raised PR #33713 for addressing the test failure.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] asfgit closed pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
asfgit closed pull request #33613:
URL: https://github.com/apache/spark/pull/33613
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Victsm commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
Victsm commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r682952969
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -471,9 +488,10 @@ public void onData(String streamId, ByteBuffer buf) {
public void onComplete(String streamId) {
if (isStaleBlockOrTooLate) {
// Throw an exception here so the block data is drained from channel and server
- // responds RpcFailure to the client.
- throw new RuntimeException(String.format("Block %s %s", streamId,
- ErrorHandler.BlockPushErrorHandler.TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX));
+ // responds the error code to the client.
+ throw new BlockPushNonFatalFailure(
+ new PushBlockNonFatalErrorCode(ErrorCode.TOO_LATE_OR_STALE_BLOCK_PUSH.id())
Review comment:
@otterc you are right that we can just replace `StaleBlockPushException` with `BlockPushNonFatalFailure`. Catching it and later throwing another is not necessary.
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -471,9 +488,10 @@ public void onData(String streamId, ByteBuffer buf) {
public void onComplete(String streamId) {
if (isStaleBlockOrTooLate) {
// Throw an exception here so the block data is drained from channel and server
- // responds RpcFailure to the client.
- throw new RuntimeException(String.format("Block %s %s", streamId,
- ErrorHandler.BlockPushErrorHandler.TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX));
+ // responds the error code to the client.
+ throw new BlockPushNonFatalFailure(
+ new PushBlockNonFatalErrorCode(ErrorCode.TOO_LATE_OR_STALE_BLOCK_PUSH.id())
Review comment:
~~@otterc you are right that we can just replace `StaleBlockPushException` with `BlockPushNonFatalFailure`. Catching it and later throwing another is not necessary.~~
There is still difference between throwing the exception inside `receiveBlockDataAsStream` vs throwing it in the `StreamCallback`.
In `TransportRequestHandler#processStreamUpload`, catching an exception while trying to create the `StreamCallback` would lead to server closing the corresponding channel.
Since the channel might be used on the client side for pushing non-stale blocks as well, it's not good to just close the channel whenever the server sees a stale push.
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -471,9 +488,10 @@ public void onData(String streamId, ByteBuffer buf) {
public void onComplete(String streamId) {
if (isStaleBlockOrTooLate) {
// Throw an exception here so the block data is drained from channel and server
- // responds RpcFailure to the client.
- throw new RuntimeException(String.format("Block %s %s", streamId,
- ErrorHandler.BlockPushErrorHandler.TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX));
+ // responds the error code to the client.
+ throw new BlockPushNonFatalFailure(
+ new PushBlockNonFatalErrorCode(ErrorCode.TOO_LATE_OR_STALE_BLOCK_PUSH.id())
Review comment:
For stale push from prior app attempt, it currently indeed throws an IllegalArgumentException inside `receiveBlockDataAsStream` which would lead to server closing channels.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Victsm commented on pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
Victsm commented on pull request #33613:
URL: https://github.com/apache/spark/pull/33613#issuecomment-895413846
@Ngone51 want to see if you have additional comments or thoughts on this PR.
Since @gengliangwang plans to cut 3.2 RC tomorrow, would be good to have this patch merged before that.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Victsm commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
Victsm commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r683694326
##########
File path: common/network-common/src/main/java/org/apache/spark/network/server/BlockPushNonFatalFailure.java
##########
@@ -0,0 +1,114 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.network.server;
+
+import java.nio.ByteBuffer;
+
+/**
+ * A special RuntimeException thrown when shuffle service experiences a non-fatal failure
+ * with handling block push requests with push-based shuffle. Due to the best-effort nature
+ * of push-based shuffle, there are cases where the exceptions gets thrown under certain
+ * relatively common cases such as when a pushed block is received after the corresponding
+ * shuffle is merge finalized or when a pushed block experiences merge collision. Under these
+ * scenarios, we throw this special RuntimeException.
+ */
+public class BlockPushNonFatalFailure extends RuntimeException {
+
+ /**
+ * The error code of the failure, encoded as a ByteBuffer to be responded back to the client.
+ * Instead of responding a RPCFailure with the exception stack trace as the payload,
+ * which makes checking the content of the exception very tedious on the client side,
+ * we can respond a proper RPCResponse to make it more robust and efficient. This
+ * field is only set on the shuffle server side when the exception is originally generated.
+ */
+ private ByteBuffer response;
+
+ /**
+ * The error code of the failure. This field is only set on the client side when a
+ * BlockPushNonFatalFailure is recreated from the error code received from the server.
+ */
+ private ReturnCode returnCode;
+
+ public BlockPushNonFatalFailure(ByteBuffer response) {
+ this.response = response;
+ }
+
+ public BlockPushNonFatalFailure(ReturnCode returnCode) {
+ this.returnCode = returnCode;
+ }
+
+ /**
+ * Since this type of exception is used to only convey the error code, we reduce the
+ * exception initialization overhead by skipping filling the stack trace.
+ */
+ @Override
+ public synchronized Throwable fillInStackTrace() {
+ return this;
+ }
+
+ public ByteBuffer getResponse() {
+ return response;
+ }
+
+ public ReturnCode getReturnCode() {
+ return returnCode;
+ }
+
+ public enum ReturnCode {
+ /**
+ * Indicate the case of a successful merge of a pushed block.
+ */
+ SUCCESS(0),
Review comment:
Not quite clear how a marker at index `0` can help with potential incompatible evolutions.
Could you please explain it a bit more? @mridulm
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r682262016
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/OneForOneBlockPusher.java
##########
@@ -77,42 +80,55 @@ public OneForOneBlockPusher(
@Override
public void onSuccess(ByteBuffer response) {
- // On receipt of a successful block push
- listener.onBlockPushSuccess(blockId, new NioManagedBuffer(ByteBuffer.allocate(0)));
+ // If the response is non-empty, the server has responded some error code. Decode
+ // it and handle the error accordingly.
+ if (response.hasRemaining()) {
Review comment:
Shall we add a new code for the successful response? I think that would be more reboust. We can rename `PushBlockNonFatalErrorCode` to `BlockPushResponse` if we go this way.
##########
File path: common/network-common/src/main/java/org/apache/spark/network/server/BlockPushNonFatalFailure.java
##########
@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.network.server;
+
+import java.nio.ByteBuffer;
+
+/**
+ * A special RuntimeException thrown when shuffle service experiences a non-fatal failure
+ * with handling block push requests with push-based shuffle. Due to the best-effort nature
+ * of push-based shuffle, there are cases where the exceptions gets thrown under certain
+ * relatively common cases such as when a pushed block is received after the corresponding
+ * shuffle is merge finalized or when a pushed block experiences merge collision. Under these
+ * scenarios, we throw this special RuntimeException.
+ */
+public class BlockPushNonFatalFailure extends RuntimeException {
+
+ /**
+ * The error code of the failure, encoded as a ByteBuffer to be responded back to the client.
+ * Instead of responding a RPCFailure with the exception stack trace as the payload,
+ * which makes checking the content of the exception very tedious on the client side,
+ * we can respond a proper RPCResponse to make it more robust and efficient. This
+ * field is only set on the shuffle server side when the exception is originally generated.
+ */
+ private ByteBuffer response;
+
+ /**
+ * The error code of the failure. This field is only set on the client side when a
+ * BlockPushNonFatalFailure is recreated from the error code received from the server.
+ */
+ private ErrorCode errorCode;
+
+ public BlockPushNonFatalFailure(ByteBuffer response) {
+ this.response = response;
+ }
+
+ public BlockPushNonFatalFailure(ErrorCode errorCode) {
+ this.errorCode = errorCode;
+ }
+
+ /**
+ * Since this type of exception is used to only convey the error code, we reduce the
+ * exception initialization overhead by skipping filling the stack trace.
+ */
+ @Override
+ public synchronized Throwable fillInStackTrace() {
+ return this;
+ }
+
+ public ByteBuffer getResponse() {
+ return response;
+ }
Review comment:
```suggestion
public ByteBuffer toByteBuffer() {
return new PushBlockNonFatalErrorCode(this.errorCode).toByteBuffer();
}
```
And we don't need the constructor which accepts `ByteBuffer`.
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -471,9 +488,10 @@ public void onData(String streamId, ByteBuffer buf) {
public void onComplete(String streamId) {
if (isStaleBlockOrTooLate) {
// Throw an exception here so the block data is drained from channel and server
- // responds RpcFailure to the client.
- throw new RuntimeException(String.format("Block %s %s", streamId,
- ErrorHandler.BlockPushErrorHandler.TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX));
+ // responds the error code to the client.
+ throw new BlockPushNonFatalFailure(
+ new PushBlockNonFatalErrorCode(ErrorCode.TOO_LATE_OR_STALE_BLOCK_PUSH.id())
Review comment:
Shall we use the individual error codes for "too late push" and "stale push" separately?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Victsm commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Minor changes to address a few identified push based shuffle server side inefficiencies.
Posted by GitBox <gi...@apache.org>.
Victsm commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r682204586
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -472,8 +493,8 @@ public void onComplete(String streamId) {
if (isStaleBlockOrTooLate) {
// Throw an exception here so the block data is drained from channel and server
// responds RpcFailure to the client.
- throw new RuntimeException(String.format("Block %s %s", streamId,
- ErrorHandler.BlockPushErrorHandler.TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX));
+ throw new RuntimeException("Block " + streamId + " " +
Review comment:
Marking as resolved since we no longer format a string in the latest version.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33613:
URL: https://github.com/apache/spark/pull/33613#issuecomment-896427353
**[Test build #142284 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/142284/testReport)** for PR 33613 at commit [`edde291`](https://github.com/apache/spark/commit/edde291ac8458a272279e47ce54cf2c35e31de43).
* This patch **fails from timeout after a configured wait of `500m`**.
* This patch merges cleanly.
* This patch adds no public classes.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r683208602
##########
File path: common/network-common/src/main/java/org/apache/spark/network/server/BlockPushNonFatalFailure.java
##########
@@ -0,0 +1,114 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.network.server;
+
+import java.nio.ByteBuffer;
+
+/**
+ * A special RuntimeException thrown when shuffle service experiences a non-fatal failure
+ * with handling block push requests with push-based shuffle. Due to the best-effort nature
+ * of push-based shuffle, there are cases where the exceptions gets thrown under certain
+ * relatively common cases such as when a pushed block is received after the corresponding
+ * shuffle is merge finalized or when a pushed block experiences merge collision. Under these
+ * scenarios, we throw this special RuntimeException.
+ */
+public class BlockPushNonFatalFailure extends RuntimeException {
+
+ /**
+ * The error code of the failure, encoded as a ByteBuffer to be responded back to the client.
+ * Instead of responding a RPCFailure with the exception stack trace as the payload,
+ * which makes checking the content of the exception very tedious on the client side,
+ * we can respond a proper RPCResponse to make it more robust and efficient. This
+ * field is only set on the shuffle server side when the exception is originally generated.
+ */
+ private ByteBuffer response;
+
+ /**
+ * The error code of the failure. This field is only set on the client side when a
+ * BlockPushNonFatalFailure is recreated from the error code received from the server.
+ */
+ private ReturnCode returnCode;
+
+ public BlockPushNonFatalFailure(ByteBuffer response) {
+ this.response = response;
+ }
+
+ public BlockPushNonFatalFailure(ReturnCode returnCode) {
+ this.returnCode = returnCode;
+ }
+
+ /**
+ * Since this type of exception is used to only convey the error code, we reduce the
+ * exception initialization overhead by skipping filling the stack trace.
+ */
+ @Override
+ public synchronized Throwable fillInStackTrace() {
+ return this;
+ }
+
+ public ByteBuffer getResponse() {
+ return response;
+ }
+
+ public ReturnCode getReturnCode() {
+ return returnCode;
+ }
+
+ public enum ReturnCode {
+ /**
+ * Indicate the case of a successful merge of a pushed block.
+ */
+ SUCCESS(0),
Review comment:
`UNKNOWN` sgtm.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33613:
URL: https://github.com/apache/spark/pull/33613#issuecomment-896162488
**[Test build #142284 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/142284/testReport)** for PR 33613 at commit [`edde291`](https://github.com/apache/spark/commit/edde291ac8458a272279e47ce54cf2c35e31de43).
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
mridulm commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r684290557
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/OneForOneBlockPusher.java
##########
@@ -77,42 +81,55 @@ public OneForOneBlockPusher(
@Override
public void onSuccess(ByteBuffer response) {
- // On receipt of a successful block push
- listener.onBlockPushSuccess(blockId, new NioManagedBuffer(ByteBuffer.allocate(0)));
+ BlockPushReturnCode returnCode =
+ (BlockPushReturnCode) BlockTransferMessage.Decoder.fromByteBuffer(response);
+ // If the return code is not SUCCESS, the server has responded some error code. Handle
+ // the error accordingly.
+ if (returnCode.returnCode != ReturnCode.SUCCESS.id()) {
+ checkAndFailRemainingBlocks(index, new BlockPushNonFatalFailure(
+ BlockPushNonFatalFailure.getReturnCode(returnCode.returnCode)));
Review comment:
Move this as a helper static method into `BlockPushNonFatalFailure` ? It is a repeating pattern to create `BlockPushNonFatalFailure` given a `ReturnCode` id.
Something like `BlockPushNonFatalFailure.fromReturnCode`
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #33613:
URL: https://github.com/apache/spark/pull/33613#issuecomment-896428912
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/142284/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33613:
URL: https://github.com/apache/spark/pull/33613#issuecomment-894699650
Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/46703/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r683251551
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/OneForOneBlockPusher.java
##########
@@ -77,42 +81,55 @@ public OneForOneBlockPusher(
@Override
public void onSuccess(ByteBuffer response) {
- // On receipt of a successful block push
- listener.onBlockPushSuccess(blockId, new NioManagedBuffer(ByteBuffer.allocate(0)));
+ BlockPushReturnCode returnCode =
+ (BlockPushReturnCode) BlockTransferMessage.Decoder.fromByteBuffer(response);
+ // If the return code is not SUCCESS, the server has responded some error code. Handle
+ // the error accordingly.
+ if (returnCode.returnCode != ReturnCode.SUCCESS.id()) {
+ checkAndFailRemainingBlocks(index, new BlockPushNonFatalFailure(
+ BlockPushNonFatalFailure.getReturnCode(returnCode.returnCode)));
+ } else {
+ // On receipt of a successful block push
+ listener.onBlockPushSuccess(blockId, new NioManagedBuffer(ByteBuffer.allocate(0)));
+ }
}
@Override
public void onFailure(Throwable e) {
- // Since block push is best effort, i.e., if we encounter a block push failure that's still
- // retriable according to ErrorHandler (not a connection exception and the block is not too
- // late), we should not fail all remaining block pushes even though
- // RetryingBlockTransferor might consider this failure not retriable (exceeding max retry
- // count etc). The best effort nature makes block push tolerable of a partial completion.
- // Thus, we only fail the block that's actually failed in this case. Note that, on the
- // RetryingBlockTransferor side, if retry is initiated, it would still invalidate the
- // previous active retry listener, and retry pushing all outstanding blocks. However, since
- // the blocks to be pushed are preloaded into memory and the first attempt of pushing these
- // blocks might have already succeeded, retry pushing all the outstanding blocks should be
- // very cheap (on the client side, the block data is in memory; on the server side, the block
- // will be recognized as a duplicate which triggers noop handling). Here, by failing only the
- // one block that's actually failed, we are essentially preventing forwarding unnecessary
- // block push failures to the parent listener of the retry listener.
- //
- // Take the following as an example. For the common exception during block push handling,
- // i.e. block collision, it is considered as retriable by ErrorHandler but not retriable
- // by RetryingBlockTransferor. When we encounter a failure of this type, we only fail the
- // one block encountering this issue not the remaining blocks in the same batch. On the
- // RetryingBlockTransferor side, since this exception is considered as not retriable, it
- // would immediately invoke parent listener's onBlockTransferFailure. However, the remaining
- // blocks in the same batch would remain current and active and they won't be impacted by
- // this exception.
- if (PUSH_ERROR_HANDLER.shouldRetryError(e)) {
- String[] targetBlockId = Arrays.copyOfRange(blockIds, index, index + 1);
- failRemainingBlocks(targetBlockId, e);
- } else {
- String[] targetBlockId = Arrays.copyOfRange(blockIds, index, blockIds.length);
- failRemainingBlocks(targetBlockId, e);
- }
+ checkAndFailRemainingBlocks(index, e);
+ }
+ }
+
+ private void checkAndFailRemainingBlocks(int index, Throwable e) {
+ // Since block push is best effort, i.e., if we encounter a block push failure that's still
+ // retriable according to ErrorHandler (not a connection exception and the block is not too
+ // late), we should not fail all remaining block pushes even though
+ // RetryingBlockTransferor might consider this failure not retriable (exceeding max retry
+ // count etc). The best effort nature makes block push tolerable of a partial completion.
+ // Thus, we only fail the block that's actually failed in this case. Note that, on the
+ // RetryingBlockTransferor side, if retry is initiated, it would still invalidate the
+ // previous active retry listener, and retry pushing all outstanding blocks. However, since
+ // the blocks to be pushed are preloaded into memory and the first attempt of pushing these
+ // blocks might have already succeeded, retry pushing all the outstanding blocks should be
+ // very cheap (on the client side, the block data is in memory; on the server side, the block
+ // will be recognized as a duplicate which triggers noop handling). Here, by failing only the
+ // one block that's actually failed, we are essentially preventing forwarding unnecessary
+ // block push failures to the parent listener of the retry listener.
+ //
+ // Take the following as an example. For the common exception during block push handling,
+ // i.e. block collision, it is considered as retriable by ErrorHandler but not retriable
+ // by RetryingBlockTransferor. When we encounter a failure of this type, we only fail the
+ // one block encountering this issue not the remaining blocks in the same batch. On the
+ // RetryingBlockTransferor side, since this exception is considered as not retriable, it
+ // would immediately invoke parent listener's onBlockTransferFailure. However, the remaining
+ // blocks in the same batch would remain current and active and they won't be impacted by
+ // this exception.
Review comment:
Shall we log the specific error for each case? Also, including whether we'd retry the block or not.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #33613:
URL: https://github.com/apache/spark/pull/33613#issuecomment-895467356
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/46748/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Victsm commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
Victsm commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r682952969
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -471,9 +488,10 @@ public void onData(String streamId, ByteBuffer buf) {
public void onComplete(String streamId) {
if (isStaleBlockOrTooLate) {
// Throw an exception here so the block data is drained from channel and server
- // responds RpcFailure to the client.
- throw new RuntimeException(String.format("Block %s %s", streamId,
- ErrorHandler.BlockPushErrorHandler.TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX));
+ // responds the error code to the client.
+ throw new BlockPushNonFatalFailure(
+ new PushBlockNonFatalErrorCode(ErrorCode.TOO_LATE_OR_STALE_BLOCK_PUSH.id())
Review comment:
~~@otterc you are right that we can just replace `StaleBlockPushException` with `BlockPushNonFatalFailure`. Catching it and later throwing another is not necessary.~~
There is still difference between throwing the exception inside `receiveBlockDataAsStream` vs throwing it in the `StreamCallback`.
In `TransportRequestHandler#processStreamUpload`, catching an exception while trying to create the `StreamCallback` would lead to server closing the corresponding channel.
Since the channel might be used on the client side for pushing non-stale blocks as well, it's not good to just close the channel whenever the server sees a stale push.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #33613:
URL: https://github.com/apache/spark/pull/33613#issuecomment-896203510
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/46792/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] otterc commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
otterc commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r682876739
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -471,9 +488,10 @@ public void onData(String streamId, ByteBuffer buf) {
public void onComplete(String streamId) {
if (isStaleBlockOrTooLate) {
// Throw an exception here so the block data is drained from channel and server
- // responds RpcFailure to the client.
- throw new RuntimeException(String.format("Block %s %s", streamId,
- ErrorHandler.BlockPushErrorHandler.TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX));
+ // responds the error code to the client.
+ throw new BlockPushNonFatalFailure(
+ new PushBlockNonFatalErrorCode(ErrorCode.TOO_LATE_OR_STALE_BLOCK_PUSH.id())
Review comment:
I saw how it was being used. My point is that this is not needed any more. We can use `BlockPushNoneFatalFailure` instead of `StaleBlockPushException`.
Also, catching the `StaleBlockPushException` is not really required. I am pointing to the below lines. If `getOrCreateAppShufflePartitionInfo` throws a `RuntimeException`, then `receiveBlockDataAsStream` would just fail. In case of stale push, it's not really needed that a `StreamCallback` is returned.
```
try {
partitionInfoBeforeCheck = getOrCreateAppShufflePartitionInfo(appShuffleInfo, msg.shuffleId,
msg.shuffleMergeId, msg.reduceId);
} catch(StaleBlockPushException sbp) {
// Set partitionInfoBeforeCheck to null so that stale block push gets handled.
partitionInfoBeforeCheck = null;
}
```
So, this could just simplified with
```
AppShufflePartitionInfo partitionInfoBeforeCheck = getOrCreateAppShufflePartitionInfo(appShuffleInfo, msg.shuffleId,
msg.shuffleMergeId, msg.reduceId);
```
If the execution goes beyond this, then that means the push was not stale and then it is just too late.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Victsm commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Minor changes to address a few identified push based shuffle server side inefficiencies.
Posted by GitBox <gi...@apache.org>.
Victsm commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r681859197
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ErrorHandler.java
##########
@@ -105,16 +105,16 @@ public boolean shouldRetryError(Throwable t) {
return false;
}
- String errorStackTrace = Throwables.getStackTraceAsString(t);
// If the block is too late or stale block push, there is no need to retry it
- return !errorStackTrace.contains(TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX);
+ String msg = t.getMessage();
+ return !(msg != null && msg.contains(TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX));
Review comment:
Well, it is not really a protocol change.
We are just switching from RPCFailure to RPCResponse using a non-empty ByteBuffer.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] otterc commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Minor changes to address a few identified push based shuffle server side inefficiencies.
Posted by GitBox <gi...@apache.org>.
otterc commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r681392579
##########
File path: core/src/main/scala/org/apache/spark/shuffle/ShuffleBlockPusher.scala
##########
@@ -78,10 +76,10 @@ private[spark] class ShuffleBlockPusher(conf: SparkConf) extends Logging {
if (t.getCause != null && t.getCause.isInstanceOf[FileNotFoundException]) {
return false
}
- val errorStackTraceString = Throwables.getStackTraceAsString(t)
// If the block is too late or the invalid block push, there is no need to retry it
- !errorStackTraceString.contains(
- BlockPushErrorHandler.TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX)
+ val msg = t.getMessage
+ !(msg != null && msg.contains(
+ BlockPushErrorHandler.TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX))
Review comment:
Never mind, the `callback` that is added to `TranportResponseHandler` is the `BlockPushCallback`, so this is fine. So, please ignore the comment.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Victsm commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Minor changes to address a few identified push based shuffle server side inefficiencies.
Posted by GitBox <gi...@apache.org>.
Victsm commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r682204044
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ErrorHandler.java
##########
@@ -105,16 +105,16 @@ public boolean shouldRetryError(Throwable t) {
return false;
}
- String errorStackTrace = Throwables.getStackTraceAsString(t);
// If the block is too late or stale block push, there is no need to retry it
- return !errorStackTrace.contains(TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX);
+ String msg = t.getMessage();
+ return !(msg != null && msg.contains(TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX));
Review comment:
I have made the change to use a proper `RPCResponse` instead of the exception stack trace for the common block push failures.
Please review the updated change.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Victsm commented on pull request #33613: [SPARK-36378][SHUFFLE] Minor changes to address a few identified push based shuffle server side inefficiencies.
Posted by GitBox <gi...@apache.org>.
Victsm commented on pull request #33613:
URL: https://github.com/apache/spark/pull/33613#issuecomment-892283355
This PR has been extended with a slight protocol change so to communicate certain common block push failures to the client with an RPCResponse instead of exception stack traces.
This fixes the root cause of the 2nd and 3rd server side inefficiencies as well as making the block push failure handling more robust.
Would appreciate a fast review on this change so to commit the change before the planned 3.2 RC cut date on 08/10 as mentioned by @gengliangwang
cc @Ngone51 @mridulm @otterc @attilapiros @venkata91 @zhouyejoe @dongjoon-hyun
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r682269708
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/protocol/PushBlockNonFatalErrorCode.java
##########
@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.network.shuffle.protocol;
+
+import com.google.common.base.Objects;
+import io.netty.buffer.ByteBuf;
+
+import org.apache.commons.lang3.builder.ToStringBuilder;
+import org.apache.commons.lang3.builder.ToStringStyle;
Review comment:
3rd party libraries should be grouped into the same group.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] otterc commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
otterc commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r682962892
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -471,9 +488,10 @@ public void onData(String streamId, ByteBuffer buf) {
public void onComplete(String streamId) {
if (isStaleBlockOrTooLate) {
// Throw an exception here so the block data is drained from channel and server
- // responds RpcFailure to the client.
- throw new RuntimeException(String.format("Block %s %s", streamId,
- ErrorHandler.BlockPushErrorHandler.TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX));
+ // responds the error code to the client.
+ throw new BlockPushNonFatalFailure(
+ new PushBlockNonFatalErrorCode(ErrorCode.TOO_LATE_OR_STALE_BLOCK_PUSH.id())
Review comment:
@Victsm that's a good point but we can still replace the usage of `StaleBlockPushException` which is created with a message that uses `String.format` with `BlockPushNonFatalFailure`. We can still catch `BlockPushNonFatalFailure` and based on the `ErrorCode` enum differentiate between stale and too late.
Also, for stale block pushes that are for prior application attempts, isn't it closing the channel better? For clients that are pushing blocks from prior application attempt, we just want them to stop pushing all together correct?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] xkrogen commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
xkrogen commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r686168149
##########
File path: common/network-common/src/main/java/org/apache/spark/network/server/BlockPushNonFatalFailure.java
##########
@@ -0,0 +1,162 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.network.server;
+
+import java.nio.ByteBuffer;
+
+import com.google.common.base.Preconditions;
+
+/**
+ * A special RuntimeException thrown when shuffle service experiences a non-fatal failure
+ * with handling block push requests with push-based shuffle. Due to the best-effort nature
+ * of push-based shuffle, there are cases where the exceptions gets thrown under certain
+ * relatively common cases such as when a pushed block is received after the corresponding
+ * shuffle is merge finalized or when a pushed block experiences merge collision. Under these
+ * scenarios, we throw this special RuntimeException.
+ */
+public class BlockPushNonFatalFailure extends RuntimeException {
+ /**
+ * String constant used for generating exception messages indicating a block to be merged
+ * arrives too late on the server side. When we get a block push failure because of the
+ * block arrives too late, we will not retry pushing the block nor log the exception on
+ * the client side.
+ */
+ public static final String TOO_LATE_BLOCK_PUSH_MESSAGE_SUFFIX =
+ " is received after merged shuffle is finalized";
+
+ /**
+ * String constant used for generating exception messages indicating a block to be merged
+ * is a stale block push in the case of indeterminate stage retries on the server side.
+ * When we get a block push failure because of the block push being stale, we will not
+ * retry pushing the block nor log the exception on the client side.
+ */
+ public static final String STALE_BLOCK_PUSH_MESSAGE_SUFFIX =
+ " is a stale block push from an indeterminate stage retry";
+
+ /**
+ * String constant used for generating exception messages indicating the server couldn't
+ * append a block after all available attempts due to collision with other blocks belonging
+ * to the same shuffle partition. When we get a block push failure because of the block
+ * couldn't be written due to this reason, we will not log the exception on the client side.
+ */
+ public static final String BLOCK_APPEND_COLLISION_MSG_SUFFIX =
+ " experienced merge collision on the server side";
+
+ /**
+ * The error code of the failure, encoded as a ByteBuffer to be responded back to the client.
+ * Instead of responding a RPCFailure with the exception stack trace as the payload,
+ * which makes checking the content of the exception very tedious on the client side,
+ * we can respond a proper RPCResponse to make it more robust and efficient. This
+ * field is only set on the shuffle server side when the exception is originally generated.
+ */
+ private ByteBuffer response;
+
+ /**
+ * The error code of the failure. This field is only set on the client side when a
+ * BlockPushNonFatalFailure is recreated from the error code received from the server.
+ */
+ private ReturnCode returnCode;
+
+ public BlockPushNonFatalFailure(ByteBuffer response, String msg) {
+ super(msg);
+ this.response = response;
+ }
+
+ public BlockPushNonFatalFailure(ReturnCode returnCode, String msg) {
+ super(msg);
+ this.returnCode = returnCode;
+ }
+
+ /**
+ * Since this type of exception is used to only convey the error code, we reduce the
+ * exception initialization overhead by skipping filling the stack trace.
+ */
+ @Override
+ public synchronized Throwable fillInStackTrace() {
+ return this;
+ }
+
+ public ByteBuffer getResponse() {
+ // Ensure we do not invoke this method if response is not set
+ Preconditions.checkNotNull(response);
+ return response;
+ }
+
+ public ReturnCode getReturnCode() {
+ // Ensure we do not invoke this method if returnCode is not set
+ Preconditions.checkNotNull(returnCode);
+ return returnCode;
+ }
+
+ public enum ReturnCode {
+ /**
+ * Indicate the case of a successful merge of a pushed block.
+ */
+ SUCCESS(0),
+ /**
+ * Indicate a block to be merged arrives too late on the server side, i.e. after the
+ * corresponding shuffle has been merge finalized. When the client gets this code, it
+ * will not retry pushing the block.
+ */
+ TOO_LATE_BLOCK_PUSH(1),
+ /**
+ * Indicating the server couldn't append a block after all available attempts due to
+ * collision with other blocks belonging to the same shuffle partition.
+ */
+ BLOCK_APPEND_COLLISION_DETECTED(2),
+ /**
+ * Indicate a block received on the server side is a stale block push in the case of
+ * indeterminate stage retries. When the client receives this code, it will not retry
+ * pushing the block.
+ */
+ STALE_BLOCK_PUSH(3);
+
+ private final byte id;
+
+ ReturnCode(int id) {
+ assert id < 128 : "Cannot have more than 128 block push return code";
+ this.id = (byte) id;
+ }
+
+ public byte id() { return id; }
+ }
+
+ public static ReturnCode getReturnCode(byte id) {
+ switch (id) {
+ case 0: return ReturnCode.SUCCESS;
+ case 1: return ReturnCode.TOO_LATE_BLOCK_PUSH;
+ case 2: return ReturnCode.BLOCK_APPEND_COLLISION_DETECTED;
+ case 3: return ReturnCode.STALE_BLOCK_PUSH;
+ default: throw new IllegalArgumentException("Unknown block push return code: " + id);
+ }
+ }
+
+ public static String getErrorMsg(String blockId, ReturnCode errorCode) {
+ Preconditions.checkArgument(errorCode != ReturnCode.SUCCESS);
+ switch (errorCode) {
+ case STALE_BLOCK_PUSH:
+ return "Block " + blockId + STALE_BLOCK_PUSH_MESSAGE_SUFFIX;
+ case TOO_LATE_BLOCK_PUSH:
+ return "Block " + blockId + TOO_LATE_BLOCK_PUSH_MESSAGE_SUFFIX;
+ case BLOCK_APPEND_COLLISION_DETECTED:
+ return "Block " + blockId + BLOCK_APPEND_COLLISION_MSG_SUFFIX;
Review comment:
if there is a 1:1 mapping from error code to message suffix, can you just move the message suffix to be a field on the enum? Then you can just do something like `return errorCode.getMessage(blockId)` here, or `return "Block " + blockId + errorCode.getMessageSuffix()`
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
mridulm commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r683825920
##########
File path: common/network-common/src/main/java/org/apache/spark/network/server/BlockPushNonFatalFailure.java
##########
@@ -0,0 +1,114 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.network.server;
+
+import java.nio.ByteBuffer;
+
+/**
+ * A special RuntimeException thrown when shuffle service experiences a non-fatal failure
+ * with handling block push requests with push-based shuffle. Due to the best-effort nature
+ * of push-based shuffle, there are cases where the exceptions gets thrown under certain
+ * relatively common cases such as when a pushed block is received after the corresponding
+ * shuffle is merge finalized or when a pushed block experiences merge collision. Under these
+ * scenarios, we throw this special RuntimeException.
+ */
+public class BlockPushNonFatalFailure extends RuntimeException {
+
+ /**
+ * The error code of the failure, encoded as a ByteBuffer to be responded back to the client.
+ * Instead of responding a RPCFailure with the exception stack trace as the payload,
+ * which makes checking the content of the exception very tedious on the client side,
+ * we can respond a proper RPCResponse to make it more robust and efficient. This
+ * field is only set on the shuffle server side when the exception is originally generated.
+ */
+ private ByteBuffer response;
+
+ /**
+ * The error code of the failure. This field is only set on the client side when a
+ * BlockPushNonFatalFailure is recreated from the error code received from the server.
+ */
+ private ReturnCode returnCode;
+
+ public BlockPushNonFatalFailure(ByteBuffer response) {
+ this.response = response;
+ }
+
+ public BlockPushNonFatalFailure(ReturnCode returnCode) {
+ this.returnCode = returnCode;
+ }
+
+ /**
+ * Since this type of exception is used to only convey the error code, we reduce the
+ * exception initialization overhead by skipping filling the stack trace.
+ */
+ @Override
+ public synchronized Throwable fillInStackTrace() {
+ return this;
+ }
+
+ public ByteBuffer getResponse() {
+ return response;
+ }
+
+ public ReturnCode getReturnCode() {
+ return returnCode;
+ }
+
+ public enum ReturnCode {
+ /**
+ * Indicate the case of a successful merge of a pushed block.
+ */
+ SUCCESS(0),
Review comment:
Adding `UNKNOWN` allows for evolution of an enum when both client and server cannot be upgraded in sync: allowing for "older" side to fallback to treating the unsupported value as `UNKNOWN` and handle it as a generic error instead of causing deserialization failure or IndexOutOfBound exception (in case we are doing `enum.values[id]` sort of code).
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Victsm commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
Victsm commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r682279211
##########
File path: common/network-common/src/main/java/org/apache/spark/network/server/BlockPushNonFatalFailure.java
##########
@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.network.server;
+
+import java.nio.ByteBuffer;
+
+/**
+ * A special RuntimeException thrown when shuffle service experiences a non-fatal failure
+ * with handling block push requests with push-based shuffle. Due to the best-effort nature
+ * of push-based shuffle, there are cases where the exceptions gets thrown under certain
+ * relatively common cases such as when a pushed block is received after the corresponding
+ * shuffle is merge finalized or when a pushed block experiences merge collision. Under these
+ * scenarios, we throw this special RuntimeException.
+ */
+public class BlockPushNonFatalFailure extends RuntimeException {
+
+ /**
+ * The error code of the failure, encoded as a ByteBuffer to be responded back to the client.
+ * Instead of responding a RPCFailure with the exception stack trace as the payload,
+ * which makes checking the content of the exception very tedious on the client side,
+ * we can respond a proper RPCResponse to make it more robust and efficient. This
+ * field is only set on the shuffle server side when the exception is originally generated.
+ */
+ private ByteBuffer response;
+
+ /**
+ * The error code of the failure. This field is only set on the client side when a
+ * BlockPushNonFatalFailure is recreated from the error code received from the server.
+ */
+ private ErrorCode errorCode;
+
+ public BlockPushNonFatalFailure(ByteBuffer response) {
+ this.response = response;
+ }
+
+ public BlockPushNonFatalFailure(ErrorCode errorCode) {
+ this.errorCode = errorCode;
+ }
+
+ /**
+ * Since this type of exception is used to only convey the error code, we reduce the
+ * exception initialization overhead by skipping filling the stack trace.
+ */
+ @Override
+ public synchronized Throwable fillInStackTrace() {
+ return this;
+ }
+
+ public ByteBuffer getResponse() {
+ return response;
+ }
Review comment:
`PushBlockNonFatalErrorCode` is in network-shuffle module since it's an RPC message, and that's the reason it's done this way.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Victsm commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Minor changes to address a few identified push based shuffle server side inefficiencies.
Posted by GitBox <gi...@apache.org>.
Victsm commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r681828503
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ErrorHandler.java
##########
@@ -105,16 +105,16 @@ public boolean shouldRetryError(Throwable t) {
return false;
}
- String errorStackTrace = Throwables.getStackTraceAsString(t);
// If the block is too late or stale block push, there is no need to retry it
- return !errorStackTrace.contains(TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX);
+ String msg = t.getMessage();
+ return !(msg != null && msg.contains(TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX));
Review comment:
I actually thought of the same during the performance evaluation.
We are currently using the exception as some sort of RPC message to deliver these 2 potentially frequent messages, i.e. too late and block collision.
It incurs overhead on both server side and client side for initializing the exception and examine the content of the exception, as well as this issue you pointed out.
It is ideally much better to make these 2 a proper RPC response, e.g. responding an actual payload with a byte instead of an empty ByteBuffer.
We can still keep the response in the case of a success push and merge of a block as an empty ByteBuffer to reduce the cost of decoding ByteBuffer for every success block push.
This however is a protocol change, and I'm not sure we will be able to change it before the 3.2 RC date.
cc @gengliangwang
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] otterc commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
otterc commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r682337244
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -471,9 +488,10 @@ public void onData(String streamId, ByteBuffer buf) {
public void onComplete(String streamId) {
if (isStaleBlockOrTooLate) {
// Throw an exception here so the block data is drained from channel and server
- // responds RpcFailure to the client.
- throw new RuntimeException(String.format("Block %s %s", streamId,
- ErrorHandler.BlockPushErrorHandler.TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX));
+ // responds the error code to the client.
+ throw new BlockPushNonFatalFailure(
+ new PushBlockNonFatalErrorCode(ErrorCode.TOO_LATE_OR_STALE_BLOCK_PUSH.id())
Review comment:
Is there a need for `StaleBlockPushException` any more? The new `BlockPushNonFatalFailure` is a `RuntimeException` and if there are separate error codes for `too late` and `stale push` then having a `StaleBlockPushException` seems unnecessary. Can it be removed?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] otterc commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Minor changes to address a few identified push based shuffle server side inefficiencies.
Posted by GitBox <gi...@apache.org>.
otterc commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r681387486
##########
File path: core/src/main/scala/org/apache/spark/shuffle/ShuffleBlockPusher.scala
##########
@@ -78,10 +76,10 @@ private[spark] class ShuffleBlockPusher(conf: SparkConf) extends Logging {
if (t.getCause != null && t.getCause.isInstanceOf[FileNotFoundException]) {
return false
}
- val errorStackTraceString = Throwables.getStackTraceAsString(t)
// If the block is too late or the invalid block push, there is no need to retry it
- !errorStackTraceString.contains(
- BlockPushErrorHandler.TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX)
+ val msg = t.getMessage
+ !(msg != null && msg.contains(
+ BlockPushErrorHandler.TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX))
Review comment:
I think this should check the message in the cause. On the client side, the `RuntimeException` becomes the cause.
See TransportClient.uploadStream
```
RpcChannelListener listener = new RpcChannelListener(requestId, callback);
channel.writeAndFlush(new UploadStream(requestId, meta, data)).addListener(listener);
```
Here, RpcChannelListener wraps the exception with an IOException
```
@Override
void handleFailure(String errorMsg, Throwable cause) {
handler.removeRpcRequest(rpcRequestId);
callback.onFailure(new IOException(errorMsg, cause));
}
```
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33613:
URL: https://github.com/apache/spark/pull/33613#issuecomment-895407357
**[Test build #142236 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/142236/testReport)** for PR 33613 at commit [`81d70da`](https://github.com/apache/spark/commit/81d70dae1bfb86f7364bb218f6ee4627ec725fe7).
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Minor changes to address a few identified push based shuffle server side inefficiencies.
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r681530864
##########
File path: common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/ErrorHandlerSuite.java
##########
@@ -31,11 +31,11 @@
@Test
public void testErrorRetry() {
ErrorHandler.BlockPushErrorHandler pushHandler = new ErrorHandler.BlockPushErrorHandler();
- assertFalse(pushHandler.shouldRetryError(new RuntimeException(new IllegalArgumentException(
- ErrorHandler.BlockPushErrorHandler.TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX))));
Review comment:
Or, this's actually an invalid error case?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
mridulm commented on pull request #33613:
URL: https://github.com/apache/spark/pull/33613#issuecomment-896340161
Merged to master, had to resolve conflicts cherry-picking to branch-3.2
+CC @gengliangwang
Thanks for fixing this @Victsm !
Thanks for the reviews @Ngone51, @gengliangwang, @otterc, @venkata91, @zhuqi-lucas, @xkrogen
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] otterc commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
otterc commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r683558307
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -385,14 +397,24 @@ public StreamCallbackWithID receiveBlockDataAsStream(PushBlockStream msg) {
+ "with the current attempt id %s stored in shuffle service for application %s",
msg.appAttemptId, appShuffleInfo.attemptId, msg.appId));
}
+ // Use string concatenation here to avoid the overhead with String.format on every
+ // pushed block.
+ final String streamId = OneForOneBlockPusher.SHUFFLE_PUSH_BLOCK_PREFIX + "_"
+ + msg.shuffleId + "_" + msg.shuffleMergeId + "_" + msg.mapIndex + "_" + msg.reduceId;
// Retrieve merged shuffle file metadata
AppShufflePartitionInfo partitionInfoBeforeCheck;
+ boolean isStaleBlock = false;
+ boolean isTooLate = false;
try {
partitionInfoBeforeCheck = getOrCreateAppShufflePartitionInfo(appShuffleInfo, msg.shuffleId,
msg.shuffleMergeId, msg.reduceId);
- } catch(StaleBlockPushException sbp) {
+ isTooLate = partitionInfoBeforeCheck == null;
+ } catch(BlockPushNonFatalFailure bpf) {
Review comment:
Referring to the previous conversation
https://github.com/apache/spark/pull/33613#discussion_r682952969
I checked other places in `getOrCreateAppShufflePartitionInfo` where we are throwing `RuntimeException` which will just fail to return a `StreamCallback` and thus closing the connection on the server.
```
throw new RuntimeException(
String.format("Cannot initialize merged shuffle partition for appId %s shuffleId %s "
+ "shuffleMergeId %s reduceId %s", appShuffleInfo.appId, shuffleId, shuffleMergeId,
reduceId), e);
```
This is thrown when the server fails to create a merged data/index/meta file. So, in this case do we really want to close the connection? The client could be sending blocks of a different partition to this server as well and using the connection for that?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #33613:
URL: https://github.com/apache/spark/pull/33613#issuecomment-896203510
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/46792/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #33613:
URL: https://github.com/apache/spark/pull/33613#issuecomment-896324717
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/142289/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #33613:
URL: https://github.com/apache/spark/pull/33613#issuecomment-896214847
**[Test build #142289 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/142289/testReport)** for PR 33613 at commit [`52b7093`](https://github.com/apache/spark/commit/52b709356b367a8616acffec9f7f3a0eec834fce).
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] venkata91 commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
venkata91 commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r682840984
##########
File path: common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java
##########
@@ -471,9 +488,10 @@ public void onData(String streamId, ByteBuffer buf) {
public void onComplete(String streamId) {
if (isStaleBlockOrTooLate) {
// Throw an exception here so the block data is drained from channel and server
- // responds RpcFailure to the client.
- throw new RuntimeException(String.format("Block %s %s", streamId,
- ErrorHandler.BlockPushErrorHandler.TOO_LATE_OR_STALE_BLOCK_PUSH_MESSAGE_SUFFIX));
+ // responds the error code to the client.
+ throw new BlockPushNonFatalFailure(
+ new PushBlockNonFatalErrorCode(ErrorCode.TOO_LATE_OR_STALE_BLOCK_PUSH.id())
Review comment:
This is used with in `RemoteBlockPushResolver`, we need it to reject older `shuffleMergeId` block push requests. See this https://github.com/apache/spark/blob/d8169493b662acac31d3fc5e6c5051917428c974/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java#L399 This is required since we cannot return null in the `compute` as that would remove the `shuffleId` Also see this comment - https://github.com/apache/spark/pull/33034#discussion_r676098927
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a change in pull request #33613: [SPARK-36378][SHUFFLE] Switch to using RPCResponse to communicate common block push failures to the client.
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on a change in pull request #33613:
URL: https://github.com/apache/spark/pull/33613#discussion_r683257278
##########
File path: common/network-common/src/main/java/org/apache/spark/network/server/TransportRequestHandler.java
##########
@@ -213,7 +213,15 @@ public void onData(String streamId, ByteBuffer buf) throws IOException {
public void onComplete(String streamId) throws IOException {
try {
streamHandler.onComplete(streamId);
- callback.onSuccess(ByteBuffer.allocate(0));
+ callback.onSuccess(streamHandler.getCompletionResponse());
+ } catch (BlockPushNonFatalFailure ex) {
Review comment:
Shall we catch `BlockPushNonFatalFailure` inside the `onComplete`? We can have a response filed inside `streamHandler` to store the response value. In this way, `getCompletionResponse` can handle the success and error cases consistently.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org