You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@uniffle.apache.org by "yl09099 (via GitHub)" <gi...@apache.org> on 2023/08/12 15:42:30 UTC
[GitHub] [incubator-uniffle] yl09099 opened a new pull request, #1137: [#825-1] feat:Added the RPC interface for reassigning ShuffleServer
yl09099 opened a new pull request, #1137:
URL: https://github.com/apache/incubator-uniffle/pull/1137
### What changes were proposed in this pull request?
We need to provide an RPC interface for reassigning ShuffleServer, with support for excluding some ShuffleServer lists.
### Why are the changes needed?
Such an interface is required when dynamically allocating a shuffleServer list.
Fix: #825
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Add UT.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] advancedxy commented on pull request #1137: [#825][part-1] feat(spark): Add the RPC interface for reassigning ShuffleServer
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on PR #1137:
URL: https://github.com/apache/incubator-uniffle/pull/1137#issuecomment-1683260982
> I did a quick overview of this diagram, seems like that it only handles shuffle write failures, how about the shuffle read failure? In which case, some of the shuffle servers are done or unable to serving shuffle data, the shuffle read client would report `FetchFailedException` to trigger a parent stage recompute. Would you mind to elaborate a bit more on how that would be handled?
>
> And by the way, I don't think it's a good idea to throw an `FetchFailedException` when writing to uniffle server has been failed for multiple times.
Also kindly remind of these two high level questions.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] yl09099 commented on pull request #1137: [#825][part-1] feat(spark): Add the RPC interface for reassigning ShuffleServer
Posted by "yl09099 (via GitHub)" <gi...@apache.org>.
yl09099 commented on PR #1137:
URL: https://github.com/apache/incubator-uniffle/pull/1137#issuecomment-1676011853
@jerqi @advancedxy I have run mvn spotless:apply locally, why is there still a format error.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#825][part-1] feat(spark): Add the RPC interface for reassigning ShuffleServer [incubator-uniffle]
Posted by "yl09099 (via GitHub)" <gi...@apache.org>.
yl09099 commented on PR #1137:
URL: https://github.com/apache/incubator-uniffle/pull/1137#issuecomment-1751586229
> > This is in good shape. Just some minor comments.
> > @zuston Please take another look, when you have time.
>
> Sorry for the late reply, I have to say sorry again that don't have much time to review for this feature.
Please look at it for me when you have time.Please trigger the compilation for me again, I didn't change anything in Tez.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] advancedxy commented on a diff in pull request #1137: [#825][part-1] feat(spark): Add the RPC interface for reassigning ShuffleServer
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on code in PR #1137:
URL: https://github.com/apache/incubator-uniffle/pull/1137#discussion_r1293498548
##########
client/src/main/java/org/apache/uniffle/client/api/ShuffleWriteClient.java:
##########
@@ -78,6 +78,16 @@ ShuffleAssignmentsInfo getShuffleAssignments(
int assignmentShuffleServerNumber,
int estimateTaskConcurrency);
+ ShuffleAssignmentsInfo getReShuffleAssignments(
+ String appId,
+ int shuffleId,
+ int partitionNum,
+ int partitionNumPerRange,
+ Set<String> requiredTags,
+ int assignmentShuffleServerNumber,
+ int estimateTaskConcurrency,
+ Set<String> failuresShuffleServerIds);
Review Comment:
Ah, @leixm I remember that this interface would be moved to a ShuffleClient, rather a `ShuffleWriteClient`. is that still on the table?
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#825][part-1] feat(spark): Add the RPC interface for reassigning ShuffleServer [incubator-uniffle]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on PR #1137:
URL: https://github.com/apache/incubator-uniffle/pull/1137#issuecomment-1751890414
Oh, please take a look for these requested changes. The resolve operation by me is not valid for this merging. @advancedxy .
And I think we could push forward quickly for this feature, If I understand correctly, this has been applied in DIDI . cc @yl09099
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] advancedxy commented on pull request #1137: [#825][part-1] feat(spark): Add the RPC interface for reassigning ShuffleServer
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on PR #1137:
URL: https://github.com/apache/incubator-uniffle/pull/1137#issuecomment-1677303593
> ### What changes were proposed in this pull request?
> We need to provide an RPC interface for reassigning ShuffleServer, with support for excluding some ShuffleServer lists.
>
> Ⅰ. Overall objective:
>
> 1. During the shuffle write phase, the ShuffleServer reports faulty nodes and reallocates the ShuffleServer list;
> 2. Triggers a Stage level retry of SPARK. The shuffleServer node is excluded and reallocated before the retry.
>
> Ⅱ. Implementation logic diagram:
>
> 
>
> Ⅲ. As shown in the picture above:
>
> 1. During Shuffle registration, obtain the ShuffleServer list to be written through the RPC interface of a Coordinator Client by following the solid blue line step. The list is bound using ShuffleID.
> 2, the Task of Stage starts, solid steps, in accordance with the green by ShuffleManager Client RPC interface gets to be written for shuffleIdToShuffleHandleInfo ShuffleServer list;
> 2. In the Stage, if Task fails to write blocks to the ShuffleServer, press the steps in red to report ShuffleServer to FailedShuffleServerList in RSSShuffleManager through the RPC interface.
> 3. FailedShuffleServerList records the number of ShuffleServer failures. After the number of failures reaches the maximum number of retries of the Task level, follow the steps in dotted orange lines. Through the RPC interface of a Coordinator Client, obtain the list of ShuffleServer files to be written (the ShuffleServer files that fail to be written are excluded). After obtaining the list, go to Step 5 of the dotted orange line. Throwing a FetchFailed Exception triggers a stage-level retry for SPARK;
> 4. Attempt 1 is generated by the SPARK Stage level again. Pull the corresponding ShuffleServer list according to the green dotted line.
>
> ### Why are the changes needed?
> Such an interface is required when dynamically allocating a shuffleServer list.
>
> Fix: #825
>
> ### Does this PR introduce _any_ user-facing change?
> No.
>
> ### How was this patch tested?
> Add UT.
I did a quick overview of this diagram, seems like that it only handles shuffle write failures, how about the shuffle read failure? In which case, some of the shuffle servers are done or unable to serving shuffle data, the shuffle read client would report `FetchFailedException` to trigger a parent stage recompute. Would you mind to elaborate a bit more on how that would be handled?
And by the way, I don't think it's a good idea to throw an `FetchFailedException` when writing to uniffle server has been failed for multiple times.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] codecov-commenter commented on pull request #1137: [#825][part-1] feat(spark): Add the RPC interface for reassigning ShuffleServer
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #1137:
URL: https://github.com/apache/incubator-uniffle/pull/1137#issuecomment-1675975859
## [Codecov](https://app.codecov.io/gh/apache/incubator-uniffle/pull/1137?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
> Merging [#1137](https://app.codecov.io/gh/apache/incubator-uniffle/pull/1137?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) (da5dd33) into [master](https://app.codecov.io/gh/apache/incubator-uniffle/commit/1406e1fceeeedd87820033513da941964dbdbd74?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) (1406e1f) will **decrease** coverage by `0.05%`.
> The diff coverage is `4.25%`.
```diff
@@ Coverage Diff @@
## master #1137 +/- ##
============================================
- Coverage 53.70% 53.66% -0.05%
+ Complexity 2573 2110 -463
============================================
Files 391 320 -71
Lines 22314 14582 -7732
Branches 1867 1326 -541
============================================
- Hits 11984 7825 -4159
+ Misses 9622 6335 -3287
+ Partials 708 422 -286
```
| [Files Changed](https://app.codecov.io/gh/apache/incubator-uniffle/pull/1137?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [...he/uniffle/client/impl/ShuffleWriteClientImpl.java](https://app.codecov.io/gh/apache/incubator-uniffle/pull/1137?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y2xpZW50L3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS91bmlmZmxlL2NsaWVudC9pbXBsL1NodWZmbGVXcml0ZUNsaWVudEltcGwuamF2YQ==) | `34.75% <0.00%> (-1.23%)` | :arrow_down: |
| [...he/uniffle/coordinator/CoordinatorGrpcService.java](https://app.codecov.io/gh/apache/incubator-uniffle/pull/1137?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29vcmRpbmF0b3Ivc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3VuaWZmbGUvY29vcmRpbmF0b3IvQ29vcmRpbmF0b3JHcnBjU2VydmljZS5qYXZh) | `1.62% <0.00%> (-0.28%)` | :arrow_down: |
| [...assignment/PartitionBalanceAssignmentStrategy.java](https://app.codecov.io/gh/apache/incubator-uniffle/pull/1137?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29vcmRpbmF0b3Ivc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3VuaWZmbGUvY29vcmRpbmF0b3Ivc3RyYXRlZ3kvYXNzaWdubWVudC9QYXJ0aXRpb25CYWxhbmNlQXNzaWdubWVudFN0cmF0ZWd5LmphdmE=) | `56.52% <0.00%> (-39.07%)` | :arrow_down: |
| [...niffle/client/impl/grpc/CoordinatorGrpcClient.java](https://app.codecov.io/gh/apache/incubator-uniffle/pull/1137?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-aW50ZXJuYWwtY2xpZW50L3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS91bmlmZmxlL2NsaWVudC9pbXBsL2dycGMvQ29vcmRpbmF0b3JHcnBjQ2xpZW50LmphdmE=) | `0.00% <0.00%> (ø)` | |
| [...ent/request/RssGetReShuffleAssignmentsRequest.java](https://app.codecov.io/gh/apache/incubator-uniffle/pull/1137?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-aW50ZXJuYWwtY2xpZW50L3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS91bmlmZmxlL2NsaWVudC9yZXF1ZXN0L1Jzc0dldFJlU2h1ZmZsZUFzc2lnbm1lbnRzUmVxdWVzdC5qYXZh) | `0.00% <0.00%> (ø)` | |
| [...r/strategy/assignment/BasicAssignmentStrategy.java](https://app.codecov.io/gh/apache/incubator-uniffle/pull/1137?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29vcmRpbmF0b3Ivc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3VuaWZmbGUvY29vcmRpbmF0b3Ivc3RyYXRlZ3kvYXNzaWdubWVudC9CYXNpY0Fzc2lnbm1lbnRTdHJhdGVneS5qYXZh) | `62.50% <15.38%> (-32.74%)` | :arrow_down: |
| [...ache/uniffle/coordinator/SimpleClusterManager.java](https://app.codecov.io/gh/apache/incubator-uniffle/pull/1137?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29vcmRpbmF0b3Ivc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3VuaWZmbGUvY29vcmRpbmF0b3IvU2ltcGxlQ2x1c3Rlck1hbmFnZXIuamF2YQ==) | `74.22% <37.50%> (-3.31%)` | :arrow_down: |
... and [72 files with indirect coverage changes](https://app.codecov.io/gh/apache/incubator-uniffle/pull/1137/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
:mega: We’re building smart automated test selection to slash your CI/CD build times. [Learn more](https://about.codecov.io/iterative-testing/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on pull request #1137: [#825][part-1] feat(spark): Add the RPC interface for reassigning ShuffleServer
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on PR #1137:
URL: https://github.com/apache/incubator-uniffle/pull/1137#issuecomment-1676012049
> @jerqi @advancedxy I have run mvn spotless:apply locally, why is there still a format error.
You don't enable profile hadoop2.8.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on pull request #1137: [#825][part-1] feat(spark): Add the RPC interface for reassigning ShuffleServer
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on PR #1137:
URL: https://github.com/apache/incubator-uniffle/pull/1137#issuecomment-1682160758
@advancedxy @yl09099 @zuston Let us discuss this feature in the meeting next week.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] advancedxy commented on pull request #1137: [#825][part-1] feat(spark): Add the RPC interface for reassigning ShuffleServer
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on PR #1137:
URL: https://github.com/apache/incubator-uniffle/pull/1137#issuecomment-1676754354
I will take a look at this today or tomorrow.
The follow chart looks awesome on the surface.
Also cc @zuston if you have interest to review this.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on pull request #1137: [#825][part-1] feat(spark): Add the RPC interface for reassigning ShuffleServer
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on PR #1137:
URL: https://github.com/apache/incubator-uniffle/pull/1137#issuecomment-1685748786
> This is in good shape. Just some minor comments.
>
> @zuston Please take another look, when you have time.
Sorry for the late reply, I have to say sorry again that don't have much time to review for this feature.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#825][part-1] feat(spark): Add the RPC interface for reassigning ShuffleServer [incubator-uniffle]
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi merged PR #1137:
URL: https://github.com/apache/incubator-uniffle/pull/1137
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#825][part-1] feat(spark): Add the RPC interface for reassigning ShuffleServer [incubator-uniffle]
Posted by "yl09099 (via GitHub)" <gi...@apache.org>.
yl09099 closed pull request #1137: [#825][part-1] feat(spark): Add the RPC interface for reassigning ShuffleServer
URL: https://github.com/apache/incubator-uniffle/pull/1137
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] advancedxy commented on a diff in pull request #1137: [#825][part-1] feat(spark): Add the RPC interface for reassigning ShuffleServer
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on code in PR #1137:
URL: https://github.com/apache/incubator-uniffle/pull/1137#discussion_r1297233282
##########
coordinator/src/main/java/org/apache/uniffle/coordinator/CoordinatorGrpcService.java:
##########
@@ -405,4 +341,71 @@ private ServerNode toServerNode(ShuffleServerHeartBeatRequest request) {
StorageInfoUtils.fromProto(request.getStorageInfoMap()),
request.getServerId().getNettyPort());
}
+
+ @Override
+ public void getShuffleAssignments(
Review Comment:
This generates too much changes. Please modify the previous method in L102 - L166 directly
##########
client/src/main/java/org/apache/uniffle/client/api/ShuffleWriteClient.java:
##########
@@ -69,6 +69,16 @@ void reportShuffleResult(
Map<Integer, List<Long>> partitionToBlockIds,
int bitmapNum);
+ ShuffleAssignmentsInfo getShuffleAssignments(
Review Comment:
I believe a better option maybe:
``` java
default ShuffleAssignmentsInfo getShuffleAssignments(
String appId,
int shuffleId,
int partitionNum,
int partitionNumPerRange,
Set<String> requiredTags,
int assignmentShuffleServerNumber,
int estimateTaskConcurrency,
Set<String> faultyServerIds) {
throw new UnsupportedException(...);
}
```
##########
client/src/main/java/org/apache/uniffle/client/api/ShuffleWriteClient.java:
##########
@@ -69,6 +69,16 @@ void reportShuffleResult(
Map<Integer, List<Long>> partitionToBlockIds,
int bitmapNum);
+ ShuffleAssignmentsInfo getShuffleAssignments(
Review Comment:
So the changes in mapred/mapreduce/tez is not needed any more.
##########
coordinator/src/main/java/org/apache/uniffle/coordinator/SimpleClusterManager.java:
##########
@@ -242,15 +242,30 @@ public void add(ServerNode node) {
@Override
public List<ServerNode> getServerList(Set<String> requiredTags) {
+ return getServerList(requiredTags, null);
+ }
+
+ @Override
+ public List<ServerNode> getServerList(Set<String> requiredTags, Set<String> faultyServerIds) {
List<ServerNode> availableNodes = Lists.newArrayList();
for (ServerNode node : servers.values()) {
if (!ServerStatus.ACTIVE.equals(node.getStatus())) {
continue;
}
- if (!excludeNodes.contains(node.getId())
- && node.getTags().containsAll(requiredTags)
- && ServerStatus.ACTIVE.equals(node.getStatus())) {
- availableNodes.add(node);
+ // If the excludeServerNodes is null, the exclusion operation is not performed
+ if (faultyServerIds != null && !faultyServerIds.isEmpty()) {
+ if (!faultyServerIds.contains(node.getId())
+ && !excludeNodes.contains(node.getId())
+ && node.getTags().containsAll(requiredTags)
+ && ServerStatus.ACTIVE.equals(node.getStatus())) {
+ availableNodes.add(node);
+ }
+ } else {
+ if (!excludeNodes.contains(node.getId())
+ && node.getTags().containsAll(requiredTags)
+ && ServerStatus.ACTIVE.equals(node.getStatus())) {
+ availableNodes.add(node);
+ }
Review Comment:
The status check is not needed as it's already been checked in L247.
##########
coordinator/src/main/java/org/apache/uniffle/coordinator/SimpleClusterManager.java:
##########
@@ -242,15 +242,30 @@ public void add(ServerNode node) {
@Override
public List<ServerNode> getServerList(Set<String> requiredTags) {
+ return getServerList(requiredTags, null);
Review Comment:
I would prefer passing `Collections.emptySet` to avoid potential NPE and consistent with other places
##########
coordinator/src/main/java/org/apache/uniffle/coordinator/strategy/assignment/PartitionBalanceAssignmentStrategy.java:
##########
@@ -115,6 +135,7 @@ public int compare(ServerNode o1, ServerNode o2) {
});
if (nodes.isEmpty() || nodes.size() < replica) {
+ LOG.info("cishidenodeshi,{},{}", nodes.size(), replica);
Review Comment:
please remove this.
##########
coordinator/src/main/java/org/apache/uniffle/coordinator/SimpleClusterManager.java:
##########
@@ -242,15 +242,30 @@ public void add(ServerNode node) {
@Override
public List<ServerNode> getServerList(Set<String> requiredTags) {
+ return getServerList(requiredTags, null);
+ }
+
+ @Override
+ public List<ServerNode> getServerList(Set<String> requiredTags, Set<String> faultyServerIds) {
List<ServerNode> availableNodes = Lists.newArrayList();
for (ServerNode node : servers.values()) {
if (!ServerStatus.ACTIVE.equals(node.getStatus())) {
continue;
}
- if (!excludeNodes.contains(node.getId())
- && node.getTags().containsAll(requiredTags)
- && ServerStatus.ACTIVE.equals(node.getStatus())) {
- availableNodes.add(node);
+ // If the excludeServerNodes is null, the exclusion operation is not performed
+ if (faultyServerIds != null && !faultyServerIds.isEmpty()) {
+ if (!faultyServerIds.contains(node.getId())
+ && !excludeNodes.contains(node.getId())
+ && node.getTags().containsAll(requiredTags)
+ && ServerStatus.ACTIVE.equals(node.getStatus())) {
+ availableNodes.add(node);
+ }
+ } else {
+ if (!excludeNodes.contains(node.getId())
+ && node.getTags().containsAll(requiredTags)
+ && ServerStatus.ACTIVE.equals(node.getStatus())) {
+ availableNodes.add(node);
+ }
Review Comment:
let's refactor this and extract a method for it
```java
private boolean isNodeAvailable(Set<String> requiredTags, Set<String> faultyServerIds, ServerNode node) {
if (faultyServerIds != null && faultyServerIds.contains(node.getId())) {
return false;
}
return !excludeNodes.contains(node.getId()) && node.getTags().containsAll(requiredTags);
}
```
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] yl09099 commented on a diff in pull request #1137: [#825][part-1] feat(spark): Add the RPC interface for reassigning ShuffleServer
Posted by "yl09099 (via GitHub)" <gi...@apache.org>.
yl09099 commented on code in PR #1137:
URL: https://github.com/apache/incubator-uniffle/pull/1137#discussion_r1294479989
##########
proto/src/main/proto/Rss.proto:
##########
@@ -393,6 +395,21 @@ message GetShuffleServerRequest {
int32 estimateTaskConcurrency = 11;
}
+message GetReShuffleServerRequest {
+ string clientHost = 1;
+ string clientPort = 2;
+ string clientProperty = 3;
+ string applicationId = 4;
+ int32 shuffleId = 5;
+ int32 partitionNum = 6;
+ int32 partitionNumPerRange = 7;
+ int32 dataReplica = 8;
+ repeated string requireTags = 9;
+ int32 assignmentShuffleServerNumber = 10;
+ int32 estimateTaskConcurrency = 11;
+ repeated string shuffleServerIds = 12;
Review Comment:
> faultyServerIds
Ok, let me change that. In the back of the PR ReportFaultyShuffleServers interface
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#825][part-1] feat(spark): Add the RPC interface for reassigning ShuffleServer [incubator-uniffle]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #1137:
URL: https://github.com/apache/incubator-uniffle/pull/1137#discussion_r1349607369
##########
client/src/main/java/org/apache/uniffle/client/api/ShuffleWriteClient.java:
##########
@@ -69,6 +69,19 @@ void reportShuffleResult(
Map<Integer, List<Long>> partitionToBlockIds,
int bitmapNum);
+ default ShuffleAssignmentsInfo getShuffleAssignments(
+ String appId,
+ int shuffleId,
+ int partitionNum,
+ int partitionNumPerRange,
+ Set<String> requiredTags,
+ int assignmentShuffleServerNumber,
+ int estimateTaskConcurrency,
+ Set<String> faultyServerIds) {
+ throw new UnsupportedOperationException(
+ "Currently only SPARK2 or SPARK3 supports this method.");
Review Comment:
cc @yl09099 please use another PR to optimize this. I will merge this.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#825][part-1] feat(spark): Add the RPC interface for reassigning ShuffleServer [incubator-uniffle]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #1137:
URL: https://github.com/apache/incubator-uniffle/pull/1137#discussion_r1349607530
##########
client/src/main/java/org/apache/uniffle/client/api/ShuffleWriteClient.java:
##########
@@ -78,6 +78,16 @@ ShuffleAssignmentsInfo getShuffleAssignments(
int assignmentShuffleServerNumber,
int estimateTaskConcurrency);
+ ShuffleAssignmentsInfo getReShuffleAssignments(
+ String appId,
+ int shuffleId,
+ int partitionNum,
+ int partitionNumPerRange,
+ Set<String> requiredTags,
+ int assignmentShuffleServerNumber,
+ int estimateTaskConcurrency,
+ Set<String> failuresShuffleServerIds);
Review Comment:
I have to close this because the merging is blocked. Feel free to comment if you have any ideas
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#825][part-1] feat(spark): Add the RPC interface for reassigning ShuffleServer [incubator-uniffle]
Posted by "yl09099 (via GitHub)" <gi...@apache.org>.
yl09099 commented on PR #1137:
URL: https://github.com/apache/incubator-uniffle/pull/1137#issuecomment-1752202321
> Oh, please take a look for these requested changes. The resolve operation by me is not valid for this merging. @advancedxy .
>
> And I think we could push forward quickly for this feature, If I understand correctly, this has been applied in DIDI . cc @yl09099
OK,I'll merge as soon as possible.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] advancedxy commented on a diff in pull request #1137: [#825][part-1] feat(spark): Add the RPC interface for reassigning ShuffleServer
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on code in PR #1137:
URL: https://github.com/apache/incubator-uniffle/pull/1137#discussion_r1293472406
##########
proto/src/main/proto/Rss.proto:
##########
@@ -393,6 +395,21 @@ message GetShuffleServerRequest {
int32 estimateTaskConcurrency = 11;
}
+message GetReShuffleServerRequest {
+ string clientHost = 1;
+ string clientPort = 2;
+ string clientProperty = 3;
+ string applicationId = 4;
+ int32 shuffleId = 5;
+ int32 partitionNum = 6;
+ int32 partitionNumPerRange = 7;
+ int32 dataReplica = 8;
+ repeated string requireTags = 9;
+ int32 assignmentShuffleServerNumber = 10;
+ int32 estimateTaskConcurrency = 11;
+ repeated string shuffleServerIds = 12;
Review Comment:
I think a better name would be `excludeServerIds` or `faultyServerIds`. And I think we can simply merge this interface with `GetShuffleServerRequest`.
Also, I think it might be valuable to add a new interface, such as `ReportFaultyShuffleServers` from AM/driver to coordinator so that coordinator could decide whether which shuffle servers to exclude when a few AMs/drivers repeatedly report some shuffle servers are faulty.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] advancedxy commented on a diff in pull request #1137: [#825][part-1] feat(spark): Add the RPC interface for reassigning ShuffleServer
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on code in PR #1137:
URL: https://github.com/apache/incubator-uniffle/pull/1137#discussion_r1293484922
##########
coordinator/src/main/java/org/apache/uniffle/coordinator/strategy/assignment/BasicAssignmentStrategy.java:
##########
@@ -66,8 +66,42 @@ public PartitionRangeAssignment assign(
return new PartitionRangeAssignment(assignments);
}
- private List<ServerNode> getRequiredServers(Set<String> requiredTags, int expectedNum) {
- List<ServerNode> servers = clusterManager.getServerList(requiredTags);
+ @Override
+ public PartitionRangeAssignment assign(
+ int totalPartitionNum,
+ int partitionNumPerRange,
+ int replica,
+ Set<String> requiredTags,
+ int requiredShuffleServerNumber,
+ int estimateTaskConcurrency,
+ Set<String> excludeServerNodes) {
+ int shuffleNodesMax = clusterManager.getShuffleNodesMax();
+ int expectedShuffleNodesNum = shuffleNodesMax;
+ if (requiredShuffleServerNumber < shuffleNodesMax && requiredShuffleServerNumber > 0) {
+ expectedShuffleNodesNum = requiredShuffleServerNumber;
+ }
+ List<ServerNode> servers =
+ getRequiredServers(requiredTags, expectedShuffleNodesNum, excludeServerNodes);
+ if (servers.isEmpty() || servers.size() < replica) {
+ return new PartitionRangeAssignment(null);
+ }
+
+ SortedMap<PartitionRange, List<ServerNode>> assignments =
+ getPartitionAssignment(
+ totalPartitionNum, partitionNumPerRange, replica, servers, estimateTaskConcurrency);
+
+ return new PartitionRangeAssignment(assignments);
+ }
Review Comment:
There's a lot code duplication.
Method should simply call the variant version only
```java
public PartitionRangeAssignment assign(
int totalPartitionNum,
int partitionNumPerRange,
int replica,
Set<String> requiredTags,
int requiredShuffleServerNumber,
int estimateTaskConcurrency) {
return assign(...., excludedNodes = null); // or Lists.of()
}
```
##########
coordinator/src/main/java/org/apache/uniffle/coordinator/strategy/assignment/PartitionBalanceAssignmentStrategy.java:
##########
@@ -59,6 +59,93 @@ public PartitionBalanceAssignmentStrategy(ClusterManager clusterManager, Coordin
this.clusterManager = clusterManager;
}
+ @Override
+ public PartitionRangeAssignment assign(
Review Comment:
ditto
##########
integration-test/common/src/test/java/org/apache/uniffle/test/CoordinatorAssignmentTest.java:
##########
@@ -185,4 +191,35 @@ public void testReconfigureNodeMax() throws Exception {
}
Uninterruptibles.sleepUninterruptibly(2, TimeUnit.SECONDS);
}
+
+ @Test
+ public void testGetReShuffleAssignments() {
+ ShuffleWriteClientImpl shuffleWriteClient =
+ new ShuffleWriteClientImpl(ClientType.GRPC.name(), 3, 1000, 1, 1, 1, 1, true, 1, 1, 10, 10);
+ shuffleWriteClient.registerCoordinators(COORDINATOR_QUORUM);
+ Set<String> excludeServer = Sets.newConcurrentHashSet();
+ List<ShuffleServer> excludeShuffleServer =
+ shuffleServers.stream().limit(3).collect(Collectors.toList());
+ excludeShuffleServer.stream().map(ss -> ss.getId()).peek(excludeServer::add);
+ ShuffleAssignmentsInfo shuffleAssignmentsInfo =
+ shuffleWriteClient.getReShuffleAssignments(
+ "app1", 0, 10, 1, TAGS, SERVER_NUM + 10, -1, excludeServer);
+ List<ShuffleServerInfo> resultShuffle = Lists.newArrayList();
+ for (List<ShuffleServerInfo> ssis : shuffleAssignmentsInfo.getPartitionToServers().values()) {
+ resultShuffle.addAll(ssis);
+ }
+ List<String> collect =
+ resultShuffle.stream().map(ss -> ss.getId()).collect(Collectors.toList());
+ assertEquals(false, isContainsOne(excludeServer, Sets.newConcurrentHashSet(collect)));
+ }
+
+ private static boolean isContainsOne(Set<String> oneList, Set<String> otherList) {
+ for (String one : otherList) {
+ boolean contains = oneList.contains(one);
+ if (contains) {
+ return true;
+ }
+ }
+ return false;
Review Comment:
use `retainAll` instead? https://docs.oracle.com/javase/6/docs/api/java/util/Set.html#retainAll%28java.util.Collection%29
##########
client/src/main/java/org/apache/uniffle/client/api/ShuffleWriteClient.java:
##########
@@ -78,6 +78,16 @@ ShuffleAssignmentsInfo getShuffleAssignments(
int assignmentShuffleServerNumber,
int estimateTaskConcurrency);
+ ShuffleAssignmentsInfo getReShuffleAssignments(
Review Comment:
you could add a default implementation which throws an unsupported exception.
In that way, you don't have to provide impls for MR and tez.
##########
coordinator/src/main/java/org/apache/uniffle/coordinator/SimpleClusterManager.java:
##########
@@ -256,6 +256,32 @@ public List<ServerNode> getServerList(Set<String> requiredTags) {
return availableNodes;
}
+ @Override
+ public List<ServerNode> getServerList(Set<String> requiredTags, Set<String> excludeServerNodes) {
+ List<ServerNode> availableNodes = Lists.newArrayList();
+ for (ServerNode node : servers.values()) {
+ if (!ServerStatus.ACTIVE.equals(node.getStatus())) {
+ continue;
+ }
+ // If the excludeServerNodes is null, the exclusion operation is not performed
+ if (excludeServerNodes != null) {
+ if (!excludeServerNodes.contains(node.getId())
+ && !excludeNodes.contains(node.getId())
+ && node.getTags().containsAll(requiredTags)
+ && ServerStatus.ACTIVE.equals(node.getStatus())) {
+ availableNodes.add(node);
+ }
+ } else {
+ if (!excludeNodes.contains(node.getId())
+ && node.getTags().containsAll(requiredTags)
+ && ServerStatus.ACTIVE.equals(node.getStatus())) {
+ availableNodes.add(node);
+ }
+ }
+ }
+ return availableNodes;
Review Comment:
Code duplication.
It would be nice simpler to just call
```java
List<ServerNode> availableNodes = getServerList(requiredTags);
if (Objects.isEmpty(excludeServerNodes) {
return availableNodes;
} else {
return availableNodes.stream().filter(!excludeServerNodes::contains).toCollector(...)
}
```
##########
coordinator/src/main/java/org/apache/uniffle/coordinator/CoordinatorGrpcService.java:
##########
@@ -405,4 +407,71 @@ private ServerNode toServerNode(ShuffleServerHeartBeatRequest request) {
StorageInfoUtils.fromProto(request.getStorageInfoMap()),
request.getServerId().getNettyPort());
}
+
+ @Override
+ public void getReShuffleAssignments(
Review Comment:
Like commented in other places, this interface might be merged.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] advancedxy commented on a diff in pull request #1137: [#825][part-1] feat(spark): Add the RPC interface for reassigning ShuffleServer
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on code in PR #1137:
URL: https://github.com/apache/incubator-uniffle/pull/1137#discussion_r1297927789
##########
internal-client/src/main/java/org/apache/uniffle/client/request/RssGetShuffleAssignmentsRequest.java:
##########
@@ -40,7 +42,16 @@ public RssGetShuffleAssignmentsRequest(
int partitionNumPerRange,
int dataReplica,
Set<String> requiredTags) {
- this(appId, shuffleId, partitionNum, partitionNumPerRange, dataReplica, requiredTags, -1, -1);
+ this(
+ appId,
+ shuffleId,
+ partitionNum,
+ partitionNumPerRange,
+ dataReplica,
+ requiredTags,
+ -1,
+ -1,
+ Sets.newConcurrentHashSet());
Review Comment:
ditto.
##########
coordinator/src/main/java/org/apache/uniffle/coordinator/CoordinatorGrpcService.java:
##########
@@ -101,7 +102,7 @@ public void getShuffleServerNum(
@Override
public void getShuffleAssignments(
- GetShuffleServerRequest request,
+ RssProtos.GetShuffleServerRequest request,
Review Comment:
why this change? I believe we should import `import org.apache.uniffle.proto.RssProtos.GetShuffleServerRequest;` instead?
##########
coordinator/src/main/java/org/apache/uniffle/coordinator/CoordinatorGrpcService.java:
##########
@@ -111,9 +112,10 @@ public void getShuffleAssignments(
final Set<String> requiredTags = Sets.newHashSet(request.getRequireTagsList());
final int requiredShuffleServerNumber = request.getAssignmentShuffleServerNumber();
final int estimateTaskConcurrency = request.getEstimateTaskConcurrency();
+ final Set<String> faultyServerIds = new HashSet<>(request.getFaultyServerIdsList());
LOG.info(
- "Request of getShuffleAssignments for appId[{}], shuffleId[{}], partitionNum[{}], "
+ "Request of redo getShuffleAssignments for appId[{}], shuffleId[{}], partitionNum[{}], "
+ " partitionNumPerRange[{}], replica[{}], requiredTags[{}], requiredShuffleServerNumber[{}]",
appId,
Review Comment:
didn't notice this early, I don't think there's need to add `redo` in the logging.
Also, please add the description of faultyServerIds in the logging. you may print the first xx nodes
##########
client/src/main/java/org/apache/uniffle/client/api/ShuffleWriteClient.java:
##########
@@ -69,6 +69,19 @@ void reportShuffleResult(
Map<Integer, List<Long>> partitionToBlockIds,
int bitmapNum);
+ default ShuffleAssignmentsInfo getShuffleAssignments(
+ String appId,
+ int shuffleId,
+ int partitionNum,
+ int partitionNumPerRange,
+ Set<String> requiredTags,
+ int assignmentShuffleServerNumber,
+ int estimateTaskConcurrency,
+ Set<String> faultyServerIds) {
+ throw new UnsupportedOperationException(
+ "Currently only SPARK2 or SPARK3 supports this method.");
Review Comment:
Nit: a good practice would be
```
new UnsupportedOperationException(
this.getClass().getName() + " doesn't implement getShuffleAssignments with faultyServerIds");
```
##########
coordinator/src/main/java/org/apache/uniffle/coordinator/SimpleClusterManager.java:
##########
@@ -242,14 +242,18 @@ public void add(ServerNode node) {
@Override
public List<ServerNode> getServerList(Set<String> requiredTags) {
+ return getServerList(requiredTags, Sets.newConcurrentHashSet());
Review Comment:
`newHashSet` should be enough?
Why do we need a concurrent one?
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] yl09099 commented on pull request #1137: [#825][part-1] feat(spark): Add the RPC interface for reassigning ShuffleServer
Posted by "yl09099 (via GitHub)" <gi...@apache.org>.
yl09099 commented on PR #1137:
URL: https://github.com/apache/incubator-uniffle/pull/1137#issuecomment-1683317647
> > I did a quick overview of this diagram, seems like that it only handles shuffle write failures, how about the shuffle read failure? In which case, some of the shuffle servers are done or unable to serving shuffle data, the shuffle read client would report `FetchFailedException` to trigger a parent stage recompute. Would you mind to elaborate a bit more on how that would be handled?
> > And by the way, I don't think it's a good idea to throw an `FetchFailedException` when writing to uniffle server has been failed for multiple times.
>
> Also kindly remind of these two high level questions.
1、Read failure Someone has submitted the relevant PR earlier#787,There is a situation that is indeed not implemented, and there is no way to notify the upstream rewrite when the read fails, which would like to be implemented later.
2、FetchFailedException is a way to handle a read failure without intruding into SPARK code. I can't think of any other better way. Do you have any good suggestions?
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on pull request #1137: [#825][part-1] feat(spark): Add the RPC interface for reassigning ShuffleServer
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on PR #1137:
URL: https://github.com/apache/incubator-uniffle/pull/1137#issuecomment-1678592317
> And by the way, I don't think it's a good idea to throw an `FetchFailedException` when writing to uniffle server has been failed for multiple times.
For writing phase, I think the mechanism of task retry is enough. If the server is bad or out-of-service, the dynamic assignment is necessary when the spark task is retrying.
--
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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org