HDDS-15524. [DiskBalancer] Container parallel moves can overwrite pending source replica deletions (#10489).

Author: Gargi-jais11

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/diskbalancer/DiskBalancerService.java

@@ -38,8 +38,10 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Queue;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.ConcurrentSkipListMap;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
@@ -112,7 +114,8 @@ public class DiskBalancerService extends BackgroundService {
   private AtomicLong nextAvailableTime = new AtomicLong(Time.monotonicNow());
 
   private Set<ContainerID> inProgressContainers;
-  private ConcurrentSkipListMap<Long, Container> pendingDeletionContainers = new ConcurrentSkipListMap();
+  private final ConcurrentSkipListMap<Long, Queue<Container>> pendingDeletionContainers =
+      new ConcurrentSkipListMap<>();
   private static FaultInjector injector;
 
   /**
@@ -639,14 +642,16 @@ public BackgroundTaskResult call() {
         }
         if (moveSucceeded && newContainer != null) {
           // Add current old container to pendingDeletionContainers.
-          pendingDeletionContainers.put(clock.millis() + replicaDeletionDelay,
-              container);
+          long deadline = clock.millis() + replicaDeletionDelay;
+          pendingDeletionContainers
+              .computeIfAbsent(deadline, ignored -> new ConcurrentLinkedQueue<>())
+              .add(container);
           ContainerLogger.logMoveSuccess(newContainer.getContainerData(), sourceVolume,
               destVolume, containerSize, Time.monotonicNow() - startTime);
         }
         postCall(moveSucceeded && newContainer != null, startTime);
 
-        // pick one expired container from pendingDeletionContainers to delete
+        // Attempt to delete any pending-deletion buckets whose deadline has elapsed.
         tryCleanupOnePendingDeletionContainer();
       }
       return BackgroundTaskResult.EmptyTaskResult.newResult();
@@ -686,7 +691,8 @@ private void deleteContainer(Container container) {
     }
   }
 
-  private void cleanupPendingDeletionContainers() {
+  @VisibleForTesting
+  public void cleanupPendingDeletionContainers() {
     // delete all pending deletion containers before stop the service
     boolean ret;
     do {
@@ -695,18 +701,18 @@ private void cleanupPendingDeletionContainers() {
   }
 
   private boolean tryCleanupOnePendingDeletionContainer() {
-    Map.Entry<Long, Container> entry = pendingDeletionContainers.pollFirstEntry();
-    if (entry != null) {
-      if (entry.getKey() <= clock.millis()) {
-        // entry container is expired
-        deleteContainer(entry.getValue());
-        return true;
-      } else {
-        // put back the container
-        pendingDeletionContainers.put(entry.getKey(), entry.getValue());
-      }
+    // peek first, only remove when expired
+    Map.Entry<Long, Queue<Container>> entry = pendingDeletionContainers.firstEntry();
+    if (entry == null || entry.getKey() > clock.millis()) {
+      return false;
     }
-    return false;
+    if (!pendingDeletionContainers.remove(entry.getKey(), entry.getValue())) {
+      return false;
+    }
+    for (Container pending : entry.getValue()) {
+      deleteContainer(pending);
+    }
+    return true;
   }
 
   public DiskBalancerInfo getDiskBalancerInfo() {
@@ -887,4 +893,16 @@ private static void pauseInjector() {
   public void setReplicaDeletionDelay(long durationMills) {
     this.replicaDeletionDelay = durationMills;
   }
+
+  @VisibleForTesting
+  public int getPendingDeletionDeadlineCount() {
+    return pendingDeletionContainers.size();
+  }
+
+  @VisibleForTesting
+  public int getPendingDeletionQueueSize() {
+    return pendingDeletionContainers.values().stream()
+        .mapToInt(Queue::size)
+        .sum();
+  }
 }

hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/diskbalancer/TestDiskBalancerTask.java

@@ -39,6 +39,7 @@
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
+import java.util.Arrays;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
@@ -47,6 +48,10 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import org.apache.commons.io.FileUtils;
 import org.apache.hadoop.hdds.HddsConfigKeys;
@@ -58,6 +63,8 @@
 import org.apache.hadoop.hdds.scm.ScmConfigKeys;
 import org.apache.hadoop.hdds.scm.container.ContainerID;
 import org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException;
+import org.apache.hadoop.hdds.utils.BackgroundTaskQueue;
+import org.apache.hadoop.hdds.utils.BackgroundTaskResult;
 import org.apache.hadoop.hdds.utils.FaultInjector;
 import org.apache.hadoop.ozone.container.checksum.ContainerChecksumTreeManager;
 import org.apache.hadoop.ozone.container.common.helpers.ContainerMetrics;
@@ -679,6 +686,81 @@ public void testMoveSkippedWhenContainerStateChanged(State invalidState)
     assertEquals(initialSourceDelta, diskBalancerService.getDeltaSizes().get(sourceVolume));
   }
 
+  @ContainerTestVersionInfo.ContainerTest
+  public void testPendingDeletionDoesNotDropReplicasOnSameMillisecondKey(
+      ContainerTestVersionInfo versionInfo)
+      throws Exception {
+    setLayoutAndSchemaForTest(versionInfo);
+
+    long delayMs = 2_000L;
+    diskBalancerService.setReplicaDeletionDelay(delayMs);
+
+    long id1 = CONTAINER_ID;
+    long id2 = CONTAINER_ID + 1;
+    long initialSourceUsed = sourceVolume.getCurrentUsage().getUsedSpace();
+
+    Container c1 = createContainer(id1, sourceVolume, State.CLOSED);
+    Container c2 = createContainer(id2, sourceVolume, State.CLOSED);
+
+    File oldDir1 = new File(c1.getContainerData().getContainerPath());
+    File oldDir2 = new File(c2.getContainerData().getContainerPath());
+    assertTrue(oldDir1.exists());
+    assertTrue(oldDir2.exists());
+
+    // Reserve dest space like the choosing policy would.
+    destVolume.incCommittedBytes(c1.getContainerData().getBytesUsed());
+    destVolume.incCommittedBytes(c2.getContainerData().getBytesUsed());
+
+    // Schedule two moves (parallelThread default is 5 in config).
+    BackgroundTaskQueue queue = diskBalancerService.getTasks();
+    assertEquals(2, queue.size());
+    DiskBalancerService.DiskBalancerTask task1 =
+        (DiskBalancerService.DiskBalancerTask) queue.poll();
+    DiskBalancerService.DiskBalancerTask task2 =
+        (DiskBalancerService.DiskBalancerTask) queue.poll();
+    assertNotNull(task1);
+    assertNotNull(task2);
+
+    // Run both moves concurrently; fixed TestClock => same deadline key.
+    ExecutorService pool = Executors.newFixedThreadPool(2);
+    try {
+      List<Future<BackgroundTaskResult>> futures = pool.invokeAll(Arrays.asList(
+          task1::call,
+          task2::call));
+      for (Future<BackgroundTaskResult> future : futures) {
+        future.get(30, TimeUnit.SECONDS);
+      }
+    } finally {
+      pool.shutdownNow();
+    }
+
+    assertEquals(2, diskBalancerService.getMetrics().getSuccessCount());
+
+    assertEquals(1, diskBalancerService.getPendingDeletionDeadlineCount(),
+        "both moves should share one deadline key");
+    assertEquals(2, diskBalancerService.getPendingDeletionQueueSize(),
+        "both container replicas should be queued for deletion");
+
+    // Not deleted yet — delay has not elapsed.
+    assertTrue(oldDir1.exists());
+    assertTrue(oldDir2.exists());
+
+    clock.fastForward(delayMs);
+    diskBalancerService.cleanupPendingDeletionContainers();
+
+    assertEquals(0, diskBalancerService.getPendingDeletionQueueSize());
+    assertFalse(oldDir1.exists());
+    assertFalse(oldDir2.exists());
+    assertFalse(sourceVolume.getContainerIterator().hasNext(),
+        "source volume should have no containers after delayed deletion");
+    assertEquals(initialSourceUsed, sourceVolume.getCurrentUsage().getUsedSpace(),
+        "source volume used space should return to pre-move level after old replicas are deleted");
+
+    // New replicas live on dest volume.
+    assertTrue(new File(containerSet.getContainer(id1).getContainerData().getContainerPath()).exists());
+    assertTrue(new File(containerSet.getContainer(id2).getContainerData().getContainerPath()).exists());
+  }
+
   private KeyValueContainer createContainer(long containerId, HddsVolume vol, State state)
       throws IOException {
     KeyValueContainerData containerData = new KeyValueContainerData(