sunchao commented on code in PR #3745:
URL: https://github.com/apache/celeborn/pull/3745#discussion_r3530821948
##########
client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java:
##########
@@ -1952,25 +2166,142 @@ public boolean isShuffleStageEnd(int shuffleId) throws
Exception {
public ReduceFileGroups updateFileGroup(
int shuffleId, int partitionId, boolean isSegmentGranularityVisible)
throws CelebornIOException {
+ return updateFileGroup(shuffleId, partitionId, partitionId + 1,
isSegmentGranularityVisible);
+ }
+
+ public ReduceFileGroups updateFileGroup(
+ int shuffleId, int startPartition, int endPartition, boolean
isSegmentGranularityVisible)
+ throws CelebornIOException {
+ if (startPartition < 0 || endPartition < startPartition) {
+ throw new IllegalArgumentException(
+ String.format("Invalid reducer file group range [%d, %d)",
startPartition, endPartition));
+ }
+ if (startPartition == endPartition) {
+ return super.updateFileGroup(shuffleId, startPartition, endPartition);
+ }
Tuple3<ReduceFileGroups, String, Exception> fileGroupTuple =
- reduceFileGroupsMap.compute(
- shuffleId,
- (id, existsTuple) -> {
- if (existsTuple == null || existsTuple._1() == null) {
- return loadFileGroupInternal(shuffleId,
isSegmentGranularityVisible);
- } else {
- return existsTuple;
- }
- });
+ loadFileGroup(shuffleId, startPartition, endPartition,
isSegmentGranularityVisible);
if (fileGroupTuple._1() == null) {
throw new CelebornIOException(
- loadFileGroupException(shuffleId, partitionId,
(fileGroupTuple._2())),
+ loadFileGroupException(shuffleId, startPartition,
(fileGroupTuple._2())),
fileGroupTuple._3());
Review Comment:
Addressed in 9bd64bc. Range-based failures now use a range-aware overload
that reports the full `[startPartition, endPartition)` interval, while the
existing single-partition helper remains unchanged for Flink.
##########
client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala:
##########
@@ -495,10 +495,24 @@ class LifecycleManager(val appUniqueId: String, val conf:
CelebornConf) extends
case GetReducerFileGroup(
shuffleId: Int,
isSegmentGranularityVisible: Boolean,
- serdeVersion: SerdeVersion) =>
+ serdeVersion: SerdeVersion,
+ startPartition: Int,
+ endPartition: Int,
+ hasPartitionRange: Boolean,
+ omitMapAttempts: Boolean) =>
logDebug(
- s"Received GetShuffleFileGroup request for shuffleId $shuffleId,
isSegmentGranularityVisible $isSegmentGranularityVisible")
- handleGetReducerFileGroup(context, shuffleId,
isSegmentGranularityVisible, serdeVersion)
+ s"Received GetShuffleFileGroup request for shuffleId $shuffleId, " +
+ s"isSegmentGranularityVisible $isSegmentGranularityVisible, " +
+ s"partitionRange ${if (hasPartitionRange) s"[$startPartition,
$endPartition)" else "all"}")
Review Comment:
Addressed in 9bd64bc. The debug log now uses the actual
`GetReducerFileGroup` message name.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
