You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@celeborn.apache.org by an...@apache.org on 2023/04/28 03:19:39 UTC
[incubator-celeborn] branch main updated: [CELEBORN-479][FOLLOWUP] Add push task should check if loc is null (#1404)
This is an automated email from the ASF dual-hosted git repository.
angerszhuuuu pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git
The following commit(s) were added to refs/heads/main by this push:
new 8d933691a [CELEBORN-479][FOLLOWUP] Add push task should check if loc is null (#1404)
8d933691a is described below
commit 8d933691ae0c8a9a032592fa668116a1dddd1a43
Author: Angerszhuuuu <an...@gmail.com>
AuthorDate: Fri Apr 28 11:19:35 2023 +0800
[CELEBORN-479][FOLLOWUP] Add push task should check if loc is null (#1404)
---
.../celeborn/client/write/DataPushQueue.java | 22 ++++++++++++++--------
1 file changed, 14 insertions(+), 8 deletions(-)
diff --git a/client/src/main/java/org/apache/celeborn/client/write/DataPushQueue.java b/client/src/main/java/org/apache/celeborn/client/write/DataPushQueue.java
index 2506d3689..843cccbb8 100644
--- a/client/src/main/java/org/apache/celeborn/client/write/DataPushQueue.java
+++ b/client/src/main/java/org/apache/celeborn/client/write/DataPushQueue.java
@@ -99,15 +99,21 @@ public class DataPushQueue {
client.getPartitionLocation(appId, shuffleId, numMappers, numPartitions);
if (partitionLocationMap != null) {
PartitionLocation loc = partitionLocationMap.get(partitionId);
- Integer oldCapacity = workerCapacity.get(loc.hostAndPushPort());
- if (oldCapacity == null) {
- oldCapacity = maxInFlight - pushState.inflightPushes(loc.hostAndPushPort());
- workerCapacity.put(loc.hostAndPushPort(), oldCapacity);
- }
- if (oldCapacity > 0) {
- iterator.remove();
+ // According to CELEBORN-560, call rerun task and speculative task after LifecycleManager
+ // handle StageEnd will return empty PartitionLocation map, here loc can be null
+ if (loc != null) {
+ Integer oldCapacity = workerCapacity.get(loc.hostAndPushPort());
+ if (oldCapacity == null) {
+ oldCapacity = maxInFlight - pushState.inflightPushes(loc.hostAndPushPort());
+ workerCapacity.put(loc.hostAndPushPort(), oldCapacity);
+ }
+ if (oldCapacity > 0) {
+ iterator.remove();
+ tasks.add(task);
+ workerCapacity.put(loc.hostAndPushPort(), oldCapacity - 1);
+ }
+ } else {
tasks.add(task);
- workerCapacity.put(loc.hostAndPushPort(), oldCapacity - 1);
}
} else {
tasks.add(task);