Copilot commented on code in PR #3745:
URL: https://github.com/apache/celeborn/pull/3745#discussion_r3524185706
##########
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:
For range-based updateFileGroup calls, the error message is built via
loadFileGroupException(shuffleId, startPartition, ...), which formats it as
"partition %d". When the request covers multiple partitions, this message is
misleading and makes debugging harder; it should mention the full requested
range.
##########
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:
The debug log message says "GetShuffleFileGroup" but the handled RPC/message
type is GetReducerFileGroup. This mismatch can confuse troubleshooting and log
searches; align the log text with the actual 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]
