apache/celeborn
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

[CELEBORN-453] In the removeExpiredShuffle method, the taskShuffleId should be removed instead of the shuffleId (#1368)

Browse Source
This commit is contained in:
jiaoqingbo 2023-03-21 16:49:50 +08:00 committed by GitHub
parent f16c7b414e
commit ba89ee98ce
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 5 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/ShuffleTaskInfo.java
View File

@ -73,7 +73,7 @@ public class ShuffleTaskInfo {
if (shuffleIdToTaskShuffleId.containsKey(shuffleId)) {
String taskShuffleId = shuffleIdToTaskShuffleId.remove(shuffleId);
taskShuffleIdToShuffleId.remove(taskShuffleId);
taskShuffleAttemptIdIndex.remove(shuffleId);
taskShuffleAttemptIdIndex.remove(taskShuffleId);
taskShuffleAttemptIdToAttemptId.remove(taskShuffleId);
}
}

5
client-flink/flink-1.14/src/test/java/org/apache/celeborn/plugin/flink/ShuffleTaskInfoSuitJ.java
View File

@ -47,10 +47,13 @@ public class ShuffleTaskInfoSuitJ {
shuffleTaskInfo.removeExpiredShuffle(encodeShuffleId);
int encodeShuffleIdNew = shuffleTaskInfo.getShuffleId("shuffleId");
Assert.assertEquals(encodeShuffleIdNew, 2);
int encodeAttemptId211 = shuffleTaskInfo.getAttemptId("shuffleId", 1, "attempt1");
Assert.assertEquals(encodeAttemptId211, 0);
}
@Test
public void testRemoveNonExistShuffl() {
public void testRemoveNonExistShuffle() {
ShuffleTaskInfo shuffleTaskInfo = new ShuffleTaskInfo();
// remove none exist shuffle
shuffleTaskInfo.removeExpiredShuffle(0);