[#2753] fix(spark): ignore removed failed blocks

client-spark/common/src/main/java/org/apache/uniffle/shuffle/ReassignExecutor.java

@@ -133,6 +133,9 @@ private void reassignAndResendForFailedBlocks(FailedBlockSendTracker failedBlock
     for (Long blockId : failedBlockIds) {
       List<TrackingBlockStatus> failedBlockStatus =
           failedBlockSendTracker.getFailedBlockStatus(blockId);
+      if (CollectionUtils.isEmpty(failedBlockStatus)) {
+        continue;
+      }
       synchronized (failedBlockStatus) {
         int retryCnt =
             failedBlockStatus.stream()
@@ -180,6 +183,9 @@ private void reassignAndResendForFailedBlocks(FailedBlockSendTracker failedBlock
         resendBlocks.addAll(failedBlockStatus);
       }
     }
+    if (resendBlocks.isEmpty()) {
+      return;
+    }
     reassignAndResendBlocks(resendBlocks);
   }
 

client-spark/common/src/test/java/org/apache/uniffle/shuffle/ReassignExecutorTest.java

@@ -18,6 +18,7 @@
 package org.apache.uniffle.shuffle;
 
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -44,6 +45,7 @@
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
 import static org.mockito.Mockito.when;
 
 public class ReassignExecutorTest {
@@ -104,6 +106,18 @@ void testRetryExceededShouldFailAndReleaseResources() {
     verify(blockInfo).executeCompletionCallback(true);
   }
 
+  @Test
+  void testRemovedFailureStatusShouldBeIgnored() {
+    long blockId = 100L;
+    when(failedBlockSendTracker.getFailedBlockIds())
+        .thenReturn(new HashSet<>(Arrays.asList(blockId)));
+    when(failedBlockSendTracker.getFailedBlockStatus(blockId)).thenReturn(Collections.emptyList());
+
+    executor.reassign();
+
+    verifyNoInteractions(removeBlockStatsFunction, resendBlocksFunction, shuffleManagerClient);
+  }
+
   @Test
   public void testMixedSameAndDifferent() throws Exception {
     Map<Integer, Pair<List<String>, List<String>>> map = new HashMap<>();

client/src/main/java/org/apache/uniffle/client/impl/FailedBlockSendTracker.java

@@ -84,11 +84,12 @@ public void clearAndReleaseBlockResources() {
   }
 
   public Set<Long> getFailedBlockIds() {
-    return trackingBlockStatusMap.keySet();
+    return Sets.newHashSet(trackingBlockStatusMap.keySet());
   }
 
   public List<TrackingBlockStatus> getFailedBlockStatus(Long blockId) {
-    return trackingBlockStatusMap.get(blockId);
+    List<TrackingBlockStatus> statuses = trackingBlockStatusMap.get(blockId);
+    return statuses == null ? Collections.emptyList() : statuses;
   }
 
   public Set<ShuffleServerInfo> getFaultyShuffleServers() {

client/src/test/java/org/apache/uniffle/client/impl/FailedBlockSendTrackerTest.java

@@ -18,6 +18,7 @@
 package org.apache.uniffle.client.impl;
 
 import java.util.List;
+import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
@@ -30,6 +31,8 @@
 import org.apache.uniffle.common.rpc.StatusCode;
 
 import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 public class FailedBlockSendTrackerTest {
   @Test
@@ -72,4 +75,30 @@ public void test() throws Exception {
               return CollectionUtils.isEqualCollection(expected, actual);
             });
   }
+
+  @Test
+  public void getFailedBlockIdsShouldReturnSnapshot() {
+    FailedBlockSendTracker tracker = new FailedBlockSendTracker();
+    ShuffleServerInfo shuffleServerInfo = new ShuffleServerInfo("host1", 19999);
+    ShuffleBlockInfo shuffleBlockInfo =
+        new ShuffleBlockInfo(
+            0,
+            0,
+            1L,
+            0,
+            0L,
+            new byte[] {},
+            Lists.newArrayList(shuffleServerInfo),
+            0,
+            0L,
+            0L);
+
+    tracker.add(shuffleBlockInfo, shuffleServerInfo, StatusCode.INTERNAL_ERROR);
+
+    Set<Long> failedBlockIds = tracker.getFailedBlockIds();
+    tracker.remove(shuffleBlockInfo.getBlockId());
+
+    assertEquals(1, failedBlockIds.size());
+    assertTrue(tracker.getFailedBlockStatus(shuffleBlockInfo.getBlockId()).isEmpty());
+  }
 }