Skip to content

[CELEBORN-1657] Throw exception to client when exception occurred in reading sort shuffle files #2818

New issue

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

Sign up for GitHub

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

Already on GitHub? Sign in to your account

Closed
FMX wants to merge 2 commits into from
Closed

[CELEBORN-1657] Throw exception to client when exception occurred in reading sort shuffle files #2818

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
8513520
[CELEBORN-1657] Fix IndexOutOfBoundsExceptions occurred in reading sh…
FMX Oct 17, 2024
df832c4
update.
FMX Oct 17, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 21 additions & 14 deletions ...src/main/java/org/apache/celeborn/service/deploy/worker/storage/PartitionFilesSorter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,14 +22,7 @@
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
Expand All @@ -51,8 +44,12 @@
import org.slf4j.LoggerFactory;

import org.apache.celeborn.common.CelebornConf;
import org.apache.celeborn.common.exception.CelebornIOException;
import org.apache.celeborn.common.identity.UserIdentifier;
import org.apache.celeborn.common.meta.*;
import org.apache.celeborn.common.meta.DiskFileInfo;
import org.apache.celeborn.common.meta.FileInfo;
import org.apache.celeborn.common.meta.MemoryFileInfo;
import org.apache.celeborn.common.meta.ReduceFileMeta;
import org.apache.celeborn.common.metrics.source.AbstractSource;
import org.apache.celeborn.common.protocol.StorageInfo;
import org.apache.celeborn.common.unsafe.Platform;
Expand All @@ -61,11 +58,7 @@
import org.apache.celeborn.service.deploy.worker.ShuffleRecoverHelper;
import org.apache.celeborn.service.deploy.worker.WorkerSource;
import org.apache.celeborn.service.deploy.worker.memory.MemoryManager;
import org.apache.celeborn.service.deploy.worker.shuffledb.DB;
import org.apache.celeborn.service.deploy.worker.shuffledb.DBBackend;
import org.apache.celeborn.service.deploy.worker.shuffledb.DBIterator;
import org.apache.celeborn.service.deploy.worker.shuffledb.DBProvider;
import org.apache.celeborn.service.deploy.worker.shuffledb.StoreVersion;
import org.apache.celeborn.service.deploy.worker.shuffledb.*;

public class PartitionFilesSorter extends ShuffleRecoverHelper {
private static final Logger logger = LoggerFactory.getLogger(PartitionFilesSorter.class);
Expand All @@ -78,6 +71,8 @@ public class PartitionFilesSorter extends ShuffleRecoverHelper {
JavaUtils.newConcurrentHashMap();
private final ConcurrentHashMap<String, Set<String>> sortingShuffleFiles =
JavaUtils.newConcurrentHashMap();
private final ConcurrentHashMap<String, ConcurrentHashMap<String, IOException>> sortExceptions =
JavaUtils.newConcurrentHashMap();
private final Cache<String, Map<Integer, List<ShuffleBlockInfo>>> indexCache;
private final Map<String, Set<String>> indexCacheNames = JavaUtils.newConcurrentHashMap();

Expand Down Expand Up @@ -290,6 +285,11 @@ public FileInfo getSortedFileInfo(
}
}

if (sortExceptions.containsKey(shuffleKey)
&& sortExceptions.get(shuffleKey).containsKey(fileId)) {
throw sortExceptions.get(shuffleKey).get(fileId);
Copy link
Copy Markdown
Contributor

@mridulm mridulm Nov 3, 2024 •
edited
Loading

Choose a reason for hiding this comment

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

Can there be a concurrent cleanup (which can result in NPE) here ?
If yes, we should try to make this atomic, something like:

  AtomicReference<IOException> sortExceptionRef = new AtomicReference<>(null);
  sortExceptions.compute(shuffleKey,(key,  current) -> {
    if (null != current) {
      sortExceptionRef.set(current.get(fileId);
    }
    return current;
  });
  if (null != sortExceptionRef.get()) throw sortExceptionRef.get();

}

return resolve(
shuffleKey,
fileId,
Expand Down Expand Up @@ -365,6 +365,7 @@ public static void sortMemoryShuffleFile(MemoryFileInfo memoryFileInfo) {
public void cleanup(HashSet<String> expiredShuffleKeys) {
for (String expiredShuffleKey : expiredShuffleKeys) {
sortingShuffleFiles.remove(expiredShuffleKey);
sortExceptions.remove(expiredShuffleKey);
deleteSortedShuffleFiles(expiredShuffleKey);
Set<String> expiredIndexCacheItems = indexCacheNames.remove(expiredShuffleKey);
if (expiredIndexCacheItems != null) {
Expand Down Expand Up @@ -727,6 +728,9 @@ public void sort() {
singleMapIdShuffleBlockList.add(blockInfo);

index += batchHeaderLen + compressedSize;
if (compressedSize < 0) {
throw new CelebornIOException("Shuffle file " + originFilePath + " is corrupted.");
}
readBufferBySize(paddingBuf, compressedSize);
}

Expand Down Expand Up @@ -769,6 +773,9 @@ public void sort() {
} catch (Exception e) {
logger.error(
"Sorting shuffle file for " + fileId + " " + originFilePath + " failed, detail: ", e);
sortExceptions
.computeIfAbsent(shuffleKey, v -> JavaUtils.newConcurrentHashMap())
.put(fileId, new IOException(e));
} finally {
closeFiles();
Set<String> sorting = sortingShuffleFiles.get(shuffleKey);
Expand Down