[CELEBORN-2032] Create reader should change to peer by taskAttemptId #3320
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
|
cc @AngersZhuuuu @RexXiong @onebox-li PTAL. |
RexXiong
reviewed
Jun 11, 2025
| val originLocations = fileGroups.partitionGroups.get(partitionId) | ||
| val hasReplicate = originLocations.asScala.exists(p => p != null && p.hasPeer) | ||
| var locations = | ||
| if (encodedAttemptId % 2 == 1 && hasReplicate) { |
There was a problem hiding this comment.
Not all replicas of the partition location are available, Therefore hasReplicate is true not mean the partition location has peer.
AngersZhuuuu
changed the title
RexXiong
reviewed
Jun 17, 2025
| val originLocations = fileGroups.partitionGroups.get(partitionId) | ||
| val hasReplicate = originLocations.asScala.exists(p => p != null && p.hasPeer) | ||
| var locations = | ||
| if (encodedAttemptId % 2 == 1 && hasReplicate) { |
There was a problem hiding this comment.
Should we use context.attemptNumber()?
| @@ -568,6 +568,13 @@ private PartitionReader createReader( | |||
| Optional<PartitionReaderCheckpointMetadata> checkpointMetadata) | |||
| throws IOException, InterruptedException { | |||
|
|
|||
| // CELEBORN-2032. For the first creation of a reader (non-retries) and | |||
| // attemptNumber % 2 = 1, we should read the replica data first. | |||
| if (location.hasPeer() && fetchChunkRetryCnt == 0 && attemptNumber % 2 == 1) { | |||
There was a problem hiding this comment.
IMO we need remove this, as the location may already change to replica, and the Partitionlocationshould consistent with pbStreamHandler
Codecov ReportAll modified and coverable lines are covered by tests ✅
Additional details and impacted files@@ Coverage Diff @@
## main #3320 +/- ##
==========================================
- Coverage 63.54% 63.07% -0.46%
==========================================
Files 343 347 +4
Lines 20812 21296 +484
Branches 1835 1879 +44
==========================================
+ Hits 13222 13431 +209
- Misses 6630 6897 +267
- Partials 960 968 +8 ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
What changes were proposed in this pull request?
In the dual-replica scenario, when creating a reader, we should select the replica based on taskAttemptId. Usually, taskAttempt0 selects primary partitionLocation, task Attempt1 selects replica partitionLocation, and so on. This will provide better fault tolerance.
Why are the changes needed?
Since #3079, we deleted the code logic which should use replica data when task attempt is odd, but if the data of primary partitionLocation is corrupted and CelebornInputStream#fillBuffer throws exception, such as decompression failure or some other exceptions, the replica prititionLocation will not be used when the task is retried. In fact, if taskAttempt1 uses the replica partitionLocation, taskAttempt1 can run successfully.
Does this PR introduce any user-facing change?
No.
How was this patch tested?
Existing UTs.