You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@celeborn.apache.org by re...@apache.org on 2025/02/12 06:23:32 UTC
(celeborn) branch branch-0.5 updated: [CELEBORN-1846] Fix the StreamHandler usage in fetching chunk when task attempt is odd
This is an automated email from the ASF dual-hosted git repository.
rexxiong pushed a commit to branch branch-0.5
in repository https://gitbox.apache.org/repos/asf/celeborn.git
The following commit(s) were added to refs/heads/branch-0.5 by this push:
new 25f06a8a6 [CELEBORN-1846] Fix the StreamHandler usage in fetching chunk when task attempt is odd
25f06a8a6 is described below
commit 25f06a8a60e3594526ada6ed310cd546e29683f4
Author: onebox-li <ly...@163.com>
AuthorDate: Wed Feb 12 14:22:33 2025 +0800
[CELEBORN-1846] Fix the StreamHandler usage in fetching chunk when task attempt is odd
### What changes were proposed in this pull request?
The streams opened in the `streamCreatorPool` thread pool are all based on the primary locations. When the task attempt is odd, the task will start to fetch the chunk from the replica location first. This will cause using the wrong streamHandler to fetch data. To keep the logic simple, we always fetch from the primary location, and when change to peer, closing stream and use a null streamHandler when fetching peers.
### Why are the changes needed?
Avoid tasks that are slowed down by NPE and potential data problems.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Manual test.
Closes #3079 from onebox-li/fix-stream-handler.
Authored-by: onebox-li <ly...@163.com>
Signed-off-by: Shuang <lv...@alibaba-inc.com>
(cherry picked from commit f9526021c758c09a160398f9217e23d45c746c88)
Signed-off-by: Shuang <lv...@alibaba-inc.com>
---
.../celeborn/client/read/CelebornInputStream.java | 31 +++++++++++++++++-----
1 file changed, 25 insertions(+), 6 deletions(-)
diff --git a/client/src/main/java/org/apache/celeborn/client/read/CelebornInputStream.java b/client/src/main/java/org/apache/celeborn/client/read/CelebornInputStream.java
index a10bb393c..471cbce71 100644
--- a/client/src/main/java/org/apache/celeborn/client/read/CelebornInputStream.java
+++ b/client/src/main/java/org/apache/celeborn/client/read/CelebornInputStream.java
@@ -39,7 +39,9 @@ import org.apache.celeborn.client.ShuffleClient;
import org.apache.celeborn.client.compress.Decompressor;
import org.apache.celeborn.common.CelebornConf;
import org.apache.celeborn.common.exception.CelebornIOException;
+import org.apache.celeborn.common.network.client.TransportClient;
import org.apache.celeborn.common.network.client.TransportClientFactory;
+import org.apache.celeborn.common.network.protocol.TransportMessage;
import org.apache.celeborn.common.protocol.*;
import org.apache.celeborn.common.unsafe.Platform;
import org.apache.celeborn.common.util.ExceptionMaker;
@@ -335,14 +337,10 @@ public abstract class CelebornInputStream extends InputStream {
private PartitionReader createReaderWithRetry(
PartitionLocation location, PbStreamHandler pbStreamHandler) throws IOException {
- // For the first time, the location will be selected according to attemptNumber
- if (fetchChunkRetryCnt == 0 && attemptNumber % 2 == 1 && location.hasPeer()) {
- location = location.getPeer();
- logger.debug("Read peer {} for attempt {}.", location, attemptNumber);
- }
Exception lastException = null;
while (fetchChunkRetryCnt < fetchChunkMaxRetry) {
try {
+ logger.debug("Create reader for location {}", location);
if (isExcluded(location)) {
throw new CelebornIOException("Fetch data from excluded worker! " + location);
}
@@ -363,6 +361,28 @@ public abstract class CelebornInputStream extends InputStream {
fetchChunkMaxRetry,
location,
e);
+ if (pbStreamHandler != null) {
+ try {
+ TransportClient client =
+ clientFactory.createClient(location.getHost(), location.getFetchPort());
+ TransportMessage bufferStreamEnd =
+ new TransportMessage(
+ MessageType.BUFFER_STREAM_END,
+ PbBufferStreamEnd.newBuilder()
+ .setStreamType(StreamType.ChunkStream)
+ .setStreamId(pbStreamHandler.getStreamId())
+ .build()
+ .toByteArray());
+ client.sendRpc(bufferStreamEnd.toByteBuffer());
+ } catch (InterruptedException | IOException ex) {
+ logger.warn(
+ "Close {} stream {} failed",
+ location.hostAndFetchPort(),
+ pbStreamHandler.getStreamId(),
+ ex);
+ }
+ pbStreamHandler = null;
+ }
location = location.getPeer();
} else {
logger.warn(
@@ -435,7 +455,6 @@ public abstract class CelebornInputStream extends InputStream {
int fetchChunkRetryCnt,
int fetchChunkMaxRetry)
throws IOException, InterruptedException {
- logger.debug("Create reader for location {}", location);
StorageInfo storageInfo = location.getStorageInfo();
switch (storageInfo.getType()) {