You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@celeborn.apache.org by "waitinfuture (via GitHub)" <gi...@apache.org> on 2024/03/01 11:25:30 UTC
[PR] [CELEBORN-1301] Catch and throw FetchFailedException in CelebornInputStream#fillBuffer [incubator-celeborn]
waitinfuture opened a new pull request, #2349:
URL: https://github.com/apache/incubator-celeborn/pull/2349
<!--
Thanks for sending a pull request! Here are some tips for you:
- Make sure the PR title start w/ a JIRA ticket, e.g. '[CELEBORN-XXXX] Your PR title ...'.
- Be sure to keep the PR description updated to reflect all changes.
- Please write your PR title to summarize what this PR proposes.
- If possible, provide a concise example to reproduce the issue for a faster review.
-->
### What changes were proposed in this pull request?
Catch and throw FetchFailedException in CelebornInputStream#fillBuffer to enable spark's stage rerun
when fillBuffer encounters fetch chunk exception
### Why are the changes needed?
ditto
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
GA
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1301] Catch and throw FetchFailedException in CelebornInputStream#fillBuffer [incubator-celeborn]
Posted by "waitinfuture (via GitHub)" <gi...@apache.org>.
waitinfuture commented on PR #2349:
URL: https://github.com/apache/incubator-celeborn/pull/2349#issuecomment-1973220593
@mridulm please take a look again, cc @ErikFang @RexXiong
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1301] Catch and throw FetchFailedException in CelebornInputStream#fillBuffer [incubator-celeborn]
Posted by "waitinfuture (via GitHub)" <gi...@apache.org>.
waitinfuture commented on code in PR #2349:
URL: https://github.com/apache/incubator-celeborn/pull/2349#discussion_r1509043800
##########
client/src/main/java/org/apache/celeborn/client/read/CelebornInputStream.java:
##########
@@ -532,66 +559,78 @@ private boolean moveToNextChunk() throws IOException {
}
private boolean fillBuffer() throws IOException {
- if (currentChunk == null) {
- return false;
- }
-
- boolean hasData = false;
- while (currentChunk.isReadable() || moveToNextChunk()) {
- currentChunk.readBytes(sizeBuf);
- int mapId = Platform.getInt(sizeBuf, Platform.BYTE_ARRAY_OFFSET);
- int attemptId = Platform.getInt(sizeBuf, Platform.BYTE_ARRAY_OFFSET + 4);
- int batchId = Platform.getInt(sizeBuf, Platform.BYTE_ARRAY_OFFSET + 8);
- int size = Platform.getInt(sizeBuf, Platform.BYTE_ARRAY_OFFSET + 12);
-
- if (shuffleCompressionEnabled) {
- if (size > compressedBuf.length) {
- compressedBuf = new byte[size];
- }
+ try {
+ if (currentChunk == null) {
+ return false;
+ }
- currentChunk.readBytes(compressedBuf, 0, size);
- } else {
- if (size > rawDataBuf.length) {
- rawDataBuf = new byte[size];
- }
+ boolean hasData = false;
+ while (currentChunk.isReadable() || moveToNextChunk()) {
+ currentChunk.readBytes(sizeBuf);
+ int mapId = Platform.getInt(sizeBuf, Platform.BYTE_ARRAY_OFFSET);
+ int attemptId = Platform.getInt(sizeBuf, Platform.BYTE_ARRAY_OFFSET + 4);
+ int batchId = Platform.getInt(sizeBuf, Platform.BYTE_ARRAY_OFFSET + 8);
+ int size = Platform.getInt(sizeBuf, Platform.BYTE_ARRAY_OFFSET + 12);
+
+ if (shuffleCompressionEnabled) {
+ if (size > compressedBuf.length) {
+ compressedBuf = new byte[size];
+ }
- currentChunk.readBytes(rawDataBuf, 0, size);
- }
+ currentChunk.readBytes(compressedBuf, 0, size);
+ } else {
+ if (size > rawDataBuf.length) {
+ rawDataBuf = new byte[size];
+ }
- // de-duplicate
- if (attemptId == attempts[mapId]) {
- if (!batchesRead.containsKey(mapId)) {
- Set<Integer> batchSet = new HashSet<>();
- batchesRead.put(mapId, batchSet);
+ currentChunk.readBytes(rawDataBuf, 0, size);
}
- Set<Integer> batchSet = batchesRead.get(mapId);
- if (!batchSet.contains(batchId)) {
- batchSet.add(batchId);
- callback.incBytesRead(BATCH_HEADER_SIZE + size);
- if (shuffleCompressionEnabled) {
- // decompress data
- int originalLength = decompressor.getOriginalLen(compressedBuf);
- if (rawDataBuf.length < originalLength) {
- rawDataBuf = new byte[originalLength];
+
+ // de-duplicate
+ if (attemptId == attempts[mapId]) {
+ if (!batchesRead.containsKey(mapId)) {
+ Set<Integer> batchSet = new HashSet<>();
+ batchesRead.put(mapId, batchSet);
+ }
+ Set<Integer> batchSet = batchesRead.get(mapId);
+ if (!batchSet.contains(batchId)) {
+ batchSet.add(batchId);
+ callback.incBytesRead(BATCH_HEADER_SIZE + size);
+ if (shuffleCompressionEnabled) {
+ // decompress data
+ int originalLength = decompressor.getOriginalLen(compressedBuf);
+ if (rawDataBuf.length < originalLength) {
+ rawDataBuf = new byte[originalLength];
+ }
+ limit = decompressor.decompress(compressedBuf, rawDataBuf, 0);
+ } else {
+ limit = size;
}
- limit = decompressor.decompress(compressedBuf, rawDataBuf, 0);
+ position = 0;
+ hasData = true;
+ break;
} else {
- limit = size;
+ logger.debug(
+ "Skip duplicated batch: mapId {}, attemptId {}, batchId {}.",
+ mapId,
+ attemptId,
+ batchId);
}
- position = 0;
- hasData = true;
- break;
- } else {
- logger.debug(
- "Skip duplicated batch: mapId {}, attemptId {}, batchId {}.",
- mapId,
- attemptId,
- batchId);
}
}
- }
- return hasData;
+ return hasData;
+ } catch (IOException e) {
+ IOException ioe = e;
+ if (exceptionMaker != null) {
+ if (shuffleClient.reportShuffleFetchFailure(appShuffleId, shuffleId)) {
+ ioe =
+ new CelebornIOException(
Review Comment:
Not necessarily to throw FetchFailedException here, because FetchFailedException constructor automatically set fetchFailed
```
// SPARK-19276. We set the fetch failure in the task context, so that even if there is user-code
// which intercepts this exception (possibly wrapping it), the Executor can still tell there was
// a fetch failure, and send the correct error msg back to the driver. We wrap with an Option
// because the TaskContext is not defined in some test cases.
Option(TaskContext.get()).map(_.setFetchFailed(this))
```
, and Spark checks it even if the Exception doesn't match
```
case t: Throwable if hasFetchFailure && !Utils.isFatalError(t) =>
val reason = task.context.fetchFailed.get.toTaskFailedReason
if (!t.isInstanceOf[FetchFailedException]) {
// there was a fetch failure in the task, but some user code wrapped that exception
// and threw something else. Regardless, we treat it as a fetch failure.
val fetchFailedCls = classOf[FetchFailedException].getName
logWarning(s"TID ${taskId} encountered a ${fetchFailedCls} and " +
s"failed, but the ${fetchFailedCls} was hidden by another " +
s"exception. Spark is handling this like a fetch failure and ignoring the " +
s"other exception: $t")
}
setTaskFinishedAndClearInterruptStatus()
execBackend.statusUpdate(taskId, TaskState.FAILED, ser.serialize(reason))
```
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1301] Catch and throw FetchFailedException in CelebornInputStream#fillBuffer [incubator-celeborn]
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #2349:
URL: https://github.com/apache/incubator-celeborn/pull/2349#discussion_r1508874590
##########
client-spark/spark-3/src/main/scala/org/apache/spark/shuffle/celeborn/CelebornShuffleReader.scala:
##########
@@ -89,6 +92,23 @@ class CelebornShuffleReader[K, C](
}
}
+ val exceptionMaker = new ExceptionMaker() {
+ override def makeException(
+ appShuffleId: Int,
+ shuffleId: Int,
+ partitionId: Int,
+ e: Exception): Exception = {
+ new FetchFailedException(
+ null,
+ handle.shuffleId,
Review Comment:
this should be `appShuffleId` ?
##########
client-spark/spark-3/src/main/scala/org/apache/spark/shuffle/celeborn/CelebornShuffleReader.scala:
##########
@@ -89,6 +92,23 @@ class CelebornShuffleReader[K, C](
}
}
+ val exceptionMaker = new ExceptionMaker() {
+ override def makeException(
+ appShuffleId: Int,
+ shuffleId: Int,
+ partitionId: Int,
+ e: Exception): Exception = {
+ new FetchFailedException(
+ null,
+ handle.shuffleId,
Review Comment:
this should be `appShuffleId` ?
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1301] Catch and throw FetchFailedException in CelebornInputStream#fillBuffer [incubator-celeborn]
Posted by "ErikFang (via GitHub)" <gi...@apache.org>.
ErikFang commented on PR #2349:
URL: https://github.com/apache/incubator-celeborn/pull/2349#issuecomment-1975185803
having this PR, I suppose the flatMap code in
https://github.com/apache/incubator-celeborn/blob/v0.4.0-incubating/client-spark/spark-3/src/main/scala/org/apache/spark/shuffle/celeborn/CelebornShuffleReader.scala#L155
can 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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1301] Catch and throw FetchFailedException in CelebornInputStream#fillBuffer [incubator-celeborn]
Posted by "waitinfuture (via GitHub)" <gi...@apache.org>.
waitinfuture commented on code in PR #2349:
URL: https://github.com/apache/incubator-celeborn/pull/2349#discussion_r1510124998
##########
common/src/main/java/org/apache/celeborn/common/util/ExceptionMaker.java:
##########
@@ -0,0 +1,22 @@
+/*
+ * 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.celeborn.common.util;
+
+public interface ExceptionMaker {
+ Exception makeException(int appShuffleId, int shuffleId, int partitionId, Exception e);
Review Comment:
Done, thanks
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1301] Catch and throw FetchFailedException in CelebornInputStream#fillBuffer [incubator-celeborn]
Posted by "waitinfuture (via GitHub)" <gi...@apache.org>.
waitinfuture closed pull request #2349: [CELEBORN-1301] Catch and throw FetchFailedException in CelebornInputStream#fillBuffer
URL: https://github.com/apache/incubator-celeborn/pull/2349
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1301] Catch and throw FetchFailedException in CelebornInputStream#fillBuffer [incubator-celeborn]
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #2349:
URL: https://github.com/apache/incubator-celeborn/pull/2349#discussion_r1509963578
##########
common/src/main/java/org/apache/celeborn/common/util/ExceptionMaker.java:
##########
@@ -0,0 +1,22 @@
+/*
+ * 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.celeborn.common.util;
+
+public interface ExceptionMaker {
+ Exception makeException(int appShuffleId, int shuffleId, int partitionId, Exception e);
Review Comment:
Do we want to rename this and make it more specific ? like `makeFetchFailureException` or some such ?
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1301] Catch and throw FetchFailedException in CelebornInputStream#fillBuffer [incubator-celeborn]
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on PR #2349:
URL: https://github.com/apache/incubator-celeborn/pull/2349#issuecomment-1977218180
We would still need it to catch the exception and throw it, right ?
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1301] Catch and throw FetchFailedException in CelebornInputStream#fillBuffer [incubator-celeborn]
Posted by "waitinfuture (via GitHub)" <gi...@apache.org>.
waitinfuture commented on PR #2349:
URL: https://github.com/apache/incubator-celeborn/pull/2349#issuecomment-1975044407
Thanks! Merged to main(v0.5.0)/branch-0.4(v0.4.1)
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1301] Catch and throw FetchFailedException in CelebornInputStream#fillBuffer [incubator-celeborn]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #2349:
URL: https://github.com/apache/incubator-celeborn/pull/2349#issuecomment-1973093728
## [Codecov](https://app.codecov.io/gh/apache/incubator-celeborn/pull/2349?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
Attention: Patch coverage is `0%` with `1 lines` in your changes are missing coverage. Please review.
> Project coverage is 48.99%. Comparing base [(`93d2b9f`)](https://app.codecov.io/gh/apache/incubator-celeborn/commit/93d2b9f47fb69ec2ba3694df5d30a8c91c754469?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) to head [(`36793f8`)](https://app.codecov.io/gh/apache/incubator-celeborn/pull/2349?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
| [Files](https://app.codecov.io/gh/apache/incubator-celeborn/pull/2349?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...rg/apache/celeborn/common/util/ExceptionMaker.java](https://app.codecov.io/gh/apache/incubator-celeborn/pull/2349?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9jZWxlYm9ybi9jb21tb24vdXRpbC9FeGNlcHRpb25NYWtlci5qYXZh) | 0.00% | [1 Missing :warning: ](https://app.codecov.io/gh/apache/incubator-celeborn/pull/2349?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## main #2349 +/- ##
==========================================
+ Coverage 48.85% 48.99% +0.14%
==========================================
Files 207 208 +1
Lines 12965 12966 +1
Branches 1113 1113
==========================================
+ Hits 6333 6351 +18
+ Misses 6225 6211 -14
+ Partials 407 404 -3
```
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/incubator-celeborn/pull/2349?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1301] Catch and throw FetchFailedException in CelebornInputStream#fillBuffer [incubator-celeborn]
Posted by "waitinfuture (via GitHub)" <gi...@apache.org>.
waitinfuture commented on code in PR #2349:
URL: https://github.com/apache/incubator-celeborn/pull/2349#discussion_r1510124968
##########
client/src/main/java/org/apache/celeborn/client/read/CelebornInputStream.java:
##########
@@ -532,66 +559,78 @@ private boolean moveToNextChunk() throws IOException {
}
private boolean fillBuffer() throws IOException {
- if (currentChunk == null) {
- return false;
- }
-
- boolean hasData = false;
- while (currentChunk.isReadable() || moveToNextChunk()) {
- currentChunk.readBytes(sizeBuf);
- int mapId = Platform.getInt(sizeBuf, Platform.BYTE_ARRAY_OFFSET);
- int attemptId = Platform.getInt(sizeBuf, Platform.BYTE_ARRAY_OFFSET + 4);
- int batchId = Platform.getInt(sizeBuf, Platform.BYTE_ARRAY_OFFSET + 8);
- int size = Platform.getInt(sizeBuf, Platform.BYTE_ARRAY_OFFSET + 12);
-
- if (shuffleCompressionEnabled) {
- if (size > compressedBuf.length) {
- compressedBuf = new byte[size];
- }
+ try {
+ if (currentChunk == null) {
+ return false;
+ }
- currentChunk.readBytes(compressedBuf, 0, size);
- } else {
- if (size > rawDataBuf.length) {
- rawDataBuf = new byte[size];
- }
+ boolean hasData = false;
+ while (currentChunk.isReadable() || moveToNextChunk()) {
+ currentChunk.readBytes(sizeBuf);
+ int mapId = Platform.getInt(sizeBuf, Platform.BYTE_ARRAY_OFFSET);
+ int attemptId = Platform.getInt(sizeBuf, Platform.BYTE_ARRAY_OFFSET + 4);
+ int batchId = Platform.getInt(sizeBuf, Platform.BYTE_ARRAY_OFFSET + 8);
+ int size = Platform.getInt(sizeBuf, Platform.BYTE_ARRAY_OFFSET + 12);
+
+ if (shuffleCompressionEnabled) {
+ if (size > compressedBuf.length) {
+ compressedBuf = new byte[size];
+ }
- currentChunk.readBytes(rawDataBuf, 0, size);
- }
+ currentChunk.readBytes(compressedBuf, 0, size);
+ } else {
+ if (size > rawDataBuf.length) {
+ rawDataBuf = new byte[size];
+ }
- // de-duplicate
- if (attemptId == attempts[mapId]) {
- if (!batchesRead.containsKey(mapId)) {
- Set<Integer> batchSet = new HashSet<>();
- batchesRead.put(mapId, batchSet);
+ currentChunk.readBytes(rawDataBuf, 0, size);
}
- Set<Integer> batchSet = batchesRead.get(mapId);
- if (!batchSet.contains(batchId)) {
- batchSet.add(batchId);
- callback.incBytesRead(BATCH_HEADER_SIZE + size);
- if (shuffleCompressionEnabled) {
- // decompress data
- int originalLength = decompressor.getOriginalLen(compressedBuf);
- if (rawDataBuf.length < originalLength) {
- rawDataBuf = new byte[originalLength];
+
+ // de-duplicate
+ if (attemptId == attempts[mapId]) {
+ if (!batchesRead.containsKey(mapId)) {
+ Set<Integer> batchSet = new HashSet<>();
+ batchesRead.put(mapId, batchSet);
+ }
+ Set<Integer> batchSet = batchesRead.get(mapId);
+ if (!batchSet.contains(batchId)) {
+ batchSet.add(batchId);
+ callback.incBytesRead(BATCH_HEADER_SIZE + size);
+ if (shuffleCompressionEnabled) {
+ // decompress data
+ int originalLength = decompressor.getOriginalLen(compressedBuf);
+ if (rawDataBuf.length < originalLength) {
+ rawDataBuf = new byte[originalLength];
+ }
+ limit = decompressor.decompress(compressedBuf, rawDataBuf, 0);
+ } else {
+ limit = size;
}
- limit = decompressor.decompress(compressedBuf, rawDataBuf, 0);
+ position = 0;
+ hasData = true;
+ break;
} else {
- limit = size;
+ logger.debug(
+ "Skip duplicated batch: mapId {}, attemptId {}, batchId {}.",
+ mapId,
+ attemptId,
+ batchId);
}
- position = 0;
- hasData = true;
- break;
- } else {
- logger.debug(
- "Skip duplicated batch: mapId {}, attemptId {}, batchId {}.",
- mapId,
- attemptId,
- batchId);
}
}
- }
- return hasData;
+ return hasData;
+ } catch (IOException e) {
+ IOException ioe = e;
+ if (exceptionMaker != null) {
+ if (shuffleClient.reportShuffleFetchFailure(appShuffleId, shuffleId)) {
+ ioe =
+ new CelebornIOException(
Review Comment:
Done, thanks
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1301] Catch and throw FetchFailedException in CelebornInputStream#fillBuffer [incubator-celeborn]
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #2349:
URL: https://github.com/apache/incubator-celeborn/pull/2349#discussion_r1508875682
##########
client-spark/spark-3/src/main/scala/org/apache/spark/shuffle/celeborn/CelebornShuffleReader.scala:
##########
@@ -89,6 +92,23 @@ class CelebornShuffleReader[K, C](
}
}
+ val exceptionMaker = new ExceptionMaker() {
+ override def makeException(
+ appShuffleId: Int,
+ shuffleId: Int,
+ partitionId: Int,
+ e: Exception): Exception = {
+ new FetchFailedException(
+ null,
+ handle.shuffleId,
+ -1,
+ -1,
+ partitionId,
+ SparkUtils.FETCH_FAILURE_ERROR_MSG + shuffleId,
Review Comment:
Include both appShuffleId and shuffleId in message (if they are different) to help debugging ?
##########
client/src/main/java/org/apache/celeborn/client/ShuffleClient.java:
##########
@@ -210,13 +211,34 @@ public abstract void mapPartitionMapperEnd(
* @return
* @throws IOException
*/
- public abstract CelebornInputStream readPartition(
+ public CelebornInputStream readPartition(
int shuffleId,
int partitionId,
int attemptNumber,
int startMapIndex,
int endMapIndex,
MetricsCallback metricsCallback)
+ throws IOException {
+ return readPartition(
+ shuffleId,
+ partitionId,
+ attemptNumber,
+ startMapIndex,
+ endMapIndex,
+ shuffleId,
+ metricsCallback,
+ null);
Review Comment:
Incorrect parameter order ?
```suggestion
return readPartition(
shuffleId,
shuffleId,
partitionId,
attemptNumber,
startMapIndex,
endMapIndex,
metricsCallback,
null);
```
##########
client-spark/spark-3/src/main/scala/org/apache/spark/shuffle/celeborn/CelebornShuffleReader.scala:
##########
@@ -89,6 +92,23 @@ class CelebornShuffleReader[K, C](
}
}
+ val exceptionMaker = new ExceptionMaker() {
+ override def makeException(
+ appShuffleId: Int,
+ shuffleId: Int,
+ partitionId: Int,
+ e: Exception): Exception = {
+ new FetchFailedException(
+ null,
+ handle.shuffleId,
Review Comment:
This should be spark's shuffle id ?
```suggestion
appShuffleId,
```
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1301] Catch and throw FetchFailedException in CelebornInputStream#fillBuffer [incubator-celeborn]
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #2349:
URL: https://github.com/apache/incubator-celeborn/pull/2349#discussion_r1509963471
##########
client/src/main/java/org/apache/celeborn/client/read/CelebornInputStream.java:
##########
@@ -532,66 +559,78 @@ private boolean moveToNextChunk() throws IOException {
}
private boolean fillBuffer() throws IOException {
- if (currentChunk == null) {
- return false;
- }
-
- boolean hasData = false;
- while (currentChunk.isReadable() || moveToNextChunk()) {
- currentChunk.readBytes(sizeBuf);
- int mapId = Platform.getInt(sizeBuf, Platform.BYTE_ARRAY_OFFSET);
- int attemptId = Platform.getInt(sizeBuf, Platform.BYTE_ARRAY_OFFSET + 4);
- int batchId = Platform.getInt(sizeBuf, Platform.BYTE_ARRAY_OFFSET + 8);
- int size = Platform.getInt(sizeBuf, Platform.BYTE_ARRAY_OFFSET + 12);
-
- if (shuffleCompressionEnabled) {
- if (size > compressedBuf.length) {
- compressedBuf = new byte[size];
- }
+ try {
+ if (currentChunk == null) {
+ return false;
+ }
- currentChunk.readBytes(compressedBuf, 0, size);
- } else {
- if (size > rawDataBuf.length) {
- rawDataBuf = new byte[size];
- }
+ boolean hasData = false;
+ while (currentChunk.isReadable() || moveToNextChunk()) {
+ currentChunk.readBytes(sizeBuf);
+ int mapId = Platform.getInt(sizeBuf, Platform.BYTE_ARRAY_OFFSET);
+ int attemptId = Platform.getInt(sizeBuf, Platform.BYTE_ARRAY_OFFSET + 4);
+ int batchId = Platform.getInt(sizeBuf, Platform.BYTE_ARRAY_OFFSET + 8);
+ int size = Platform.getInt(sizeBuf, Platform.BYTE_ARRAY_OFFSET + 12);
+
+ if (shuffleCompressionEnabled) {
+ if (size > compressedBuf.length) {
+ compressedBuf = new byte[size];
+ }
- currentChunk.readBytes(rawDataBuf, 0, size);
- }
+ currentChunk.readBytes(compressedBuf, 0, size);
+ } else {
+ if (size > rawDataBuf.length) {
+ rawDataBuf = new byte[size];
+ }
- // de-duplicate
- if (attemptId == attempts[mapId]) {
- if (!batchesRead.containsKey(mapId)) {
- Set<Integer> batchSet = new HashSet<>();
- batchesRead.put(mapId, batchSet);
+ currentChunk.readBytes(rawDataBuf, 0, size);
}
- Set<Integer> batchSet = batchesRead.get(mapId);
- if (!batchSet.contains(batchId)) {
- batchSet.add(batchId);
- callback.incBytesRead(BATCH_HEADER_SIZE + size);
- if (shuffleCompressionEnabled) {
- // decompress data
- int originalLength = decompressor.getOriginalLen(compressedBuf);
- if (rawDataBuf.length < originalLength) {
- rawDataBuf = new byte[originalLength];
+
+ // de-duplicate
+ if (attemptId == attempts[mapId]) {
+ if (!batchesRead.containsKey(mapId)) {
+ Set<Integer> batchSet = new HashSet<>();
+ batchesRead.put(mapId, batchSet);
+ }
+ Set<Integer> batchSet = batchesRead.get(mapId);
+ if (!batchSet.contains(batchId)) {
+ batchSet.add(batchId);
+ callback.incBytesRead(BATCH_HEADER_SIZE + size);
+ if (shuffleCompressionEnabled) {
+ // decompress data
+ int originalLength = decompressor.getOriginalLen(compressedBuf);
+ if (rawDataBuf.length < originalLength) {
+ rawDataBuf = new byte[originalLength];
+ }
+ limit = decompressor.decompress(compressedBuf, rawDataBuf, 0);
+ } else {
+ limit = size;
}
- limit = decompressor.decompress(compressedBuf, rawDataBuf, 0);
+ position = 0;
+ hasData = true;
+ break;
} else {
- limit = size;
+ logger.debug(
+ "Skip duplicated batch: mapId {}, attemptId {}, batchId {}.",
+ mapId,
+ attemptId,
+ batchId);
}
- position = 0;
- hasData = true;
- break;
- } else {
- logger.debug(
- "Skip duplicated batch: mapId {}, attemptId {}, batchId {}.",
- mapId,
- attemptId,
- batchId);
}
}
- }
- return hasData;
+ return hasData;
+ } catch (IOException e) {
+ IOException ioe = e;
+ if (exceptionMaker != null) {
+ if (shuffleClient.reportShuffleFetchFailure(appShuffleId, shuffleId)) {
+ ioe =
+ new CelebornIOException(
Review Comment:
Agree, but we should add this comment to the code - it can get confusing for future code evolution otherwise :-)
Something like:
```
/*
* [[ExceptionMaker.makeException]], for spark applications with celeborn.client.spark.fetch.throwsFetchFailure enabled will result in creating
* a FetchiFailedException; and that will make the TaskContext as failed with shuffle fetch issues - see SPARK-19276 for more.
* Given this, Celeborn can wrap the FetchFailedException with our CelebornIOException
*/
```
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org