HDDS-15066. Read-Write Lock race leave stale references to container creating orphan replicas

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/ContainerSet.java

@@ -63,6 +63,12 @@ public class ContainerSet implements Iterable<Container<?>> {
 
   private static final Logger LOG = LoggerFactory.getLogger(ContainerSet.class);
 
+  /**
+   * Max attempts to acquire {@link Container#writeLock()} while verifying this set's id → container mapping
+   * is same (e.g. another thread may {@link #updateContainer} / DiskBalancer swap the instance).
+   */
+  private static final int MAX_CONTAINER_MAP_SWAP_RETRIES = 5;
+
   private final ConcurrentSkipListMap<Long, Container<?>> containerMap = new
       ConcurrentSkipListMap<>();
   private final ConcurrentSkipListSet<Long> missingContainerSet =
@@ -279,6 +285,54 @@ public Container<?> getContainer(long containerId) {
     return containerMap.get(containerId);
   }
 
+  /**
+   * Returns the max retry for a container map swap while acquiring container lock.
+   * @return max retry count
+   */
+  public static int maxContainerMapSwapRetries() {
+    return MAX_CONTAINER_MAP_SWAP_RETRIES;
+  }
+
+  /**
+   * Locks the container mapped to {@code containerId} for write, and verifies that the instance locked is still
+   * the one stored in this set. If the mapping is swapped, unlocks and retries up to
+   * {@link #maxContainerMapSwapRetries()} times, then returns {@code null}.
+   *
+   * @return the locked container, or {@code null} if the mapping could not be stabilized after all retries
+   * @throws StorageContainerException with {@code CONTAINER_NOT_FOUND}
+   */
+  @Nullable
+  public Container<?> getContainerWithWriteLock(long containerId) throws StorageContainerException {
+    for (int retry = 0; retry < MAX_CONTAINER_MAP_SWAP_RETRIES; retry++) {
+      Container<?> candidate = getContainer(containerId);
+      if (candidate == null) {
+        throw new StorageContainerException(
+            "Container " + containerId + " not found in ContainerSet.",
+            ContainerProtos.Result.CONTAINER_NOT_FOUND);
+      }
+      candidate.writeLock();
+      Container<?> current = getContainer(containerId);
+      if (current == null) {
+        candidate.writeUnlock();
+        throw new StorageContainerException(
+            "Container " + containerId + " not found in ContainerSet.",
+            ContainerProtos.Result.CONTAINER_NOT_FOUND);
+      }
+      if (current != candidate) {
+        candidate.writeUnlock();
+        if (LOG.isDebugEnabled()) {
+          LOG.debug("Container {} mapping changed during lock acquisition (attempt {}); retrying.",
+              containerId, retry);
+        }
+        continue;
+      }
+      return candidate;
+    }
+    LOG.warn("Container {} mapping kept changing after {} attempts; giving up.",
+        containerId, MAX_CONTAINER_MAP_SWAP_RETRIES);
+    return null;
+  }
+
   /**
    * Removes container from both memory and database. This should be used when the containerData on disk has been
    * removed completely from the node.

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/DeleteBlocksCommandHandler.java

@@ -303,19 +303,31 @@ public DeleteBlockTransactionExecutionResult call() {
           if (keyValueContainer.
               writeLockTryLock(tryLockTimeoutMs, TimeUnit.MILLISECONDS)) {
             try {
-              String schemaVersion = containerData
-                  .getSupportedSchemaVersionOrDefault();
-              if (getSchemaHandlers().containsKey(schemaVersion)) {
-                schemaHandlers.get(schemaVersion).handle(containerData, tx);
+              // Re-fetch the container after acquiring the lock. DiskBalancer may have relocated
+              // this container to a different disk while we waited — in that case, the container
+              // object in ContainerSet has changed and containerData points to the old replica.
+              Container<?> current = containerSet.getContainer(containerId);
+              if (current == null || current.getContainerData() != containerData) {
+                LOG.debug("DeleteBlocks: containerData for container {} is stale "
+                    + ", Will retry on the new replica.",
+                    containerId);
+                lockAcquisitionFailed = true;
+                txResultBuilder.setContainerID(containerId).setSuccess(false);
               } else {
-                throw new UnsupportedOperationException(
-                    "Only schema version 1,2,3 are supported.");
+                String schemaVersion = containerData
+                    .getSupportedSchemaVersionOrDefault();
+                if (getSchemaHandlers().containsKey(schemaVersion)) {
+                  schemaHandlers.get(schemaVersion).handle(containerData, tx);
+                } else {
+                  throw new UnsupportedOperationException(
+                      "Only schema version 1,2,3 are supported.");
+                }
+                txResultBuilder.setContainerID(containerId)
+                    .setSuccess(true);
               }
             } finally {
               keyValueContainer.writeUnlock();
             }
-            txResultBuilder.setContainerID(containerId)
-                .setSuccess(true);
           } else {
             lockAcquisitionFailed = true;
             txResultBuilder.setContainerID(containerId)

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

@@ -512,19 +512,19 @@ public BackgroundTaskResult call() {
         return BackgroundTaskResult.EmptyTaskResult.newResult();
       }
 
-      // Double check container state before acquiring lock to start move process.
-      // Container state may have changed after selection.
-      State containerState = container.getContainerData().getState();
-      if (!movableContainerStates.contains(containerState)) {
-        LOG.warn("Container {} is in {} state, skipping move process.", containerId, containerState);
-        postCall(false, startTime);
-        return BackgroundTaskResult.EmptyTaskResult.newResult();
-      }
-
       // hold read lock on the container first, to avoid other threads to update the container state,
       // such as block deletion.
       container.readLock();
       try {
+        // Double check container state after acquiring lock to start move process.
+        // Container state may have changed after selection.
+        State containerState = container.getContainerData().getState();
+        if (!movableContainerStates.contains(containerState)) {
+          LOG.warn("Container {} is in {} state, skipping move process.", containerId, containerState);
+          moveSucceeded = false;
+          return BackgroundTaskResult.EmptyTaskResult.newResult();
+        }
+
         // Step 1: Copy container to new Volume's tmp Dir
         diskBalancerTmpDir = getDiskBalancerTmpDir(destVolume)
             .resolve(String.valueOf(containerId));
@@ -580,6 +580,10 @@ public BackgroundTaskResult call() {
         // old caller can still hold the old Container object.
         ozoneContainer.getContainerSet().updateContainer(newContainer);
         destVolume.incrementUsedSpace(containerSize);
+
+        // Test injector: ContainerSet now references newContainer while this thread still holds
+        // readLock on the old replica.
+        pauseInjector();
         // Mark old container as DELETED and persist state.
         // markContainerForDelete require writeLock, so release readLock first
         container.readUnlock();
@@ -597,13 +601,7 @@ public BackgroundTaskResult call() {
         metrics.incrSuccessBytes(containerSize);
         totalBalancedBytes.addAndGet(containerSize);
       } catch (IOException e) {
-        if (injector != null) {
-          try {
-            injector.pause();
-          } catch (IOException ex) {
-            // do nothing
-          }
-        }
+        pauseInjector();
         moveSucceeded = false;
         LOG.warn("Failed to move container {}", containerId, e);
         if (diskBalancerTmpDir != null) {
@@ -861,6 +859,17 @@ public static void setInjector(FaultInjector instance) {
     injector = instance;
   }
 
+  // call FaultInjector#pause when an injector is registered; ignore IOException.
+  private static void pauseInjector() {
+    if (injector != null) {
+      try {
+        injector.pause();
+      } catch (IOException ex) {
+        // do nothing
+      }
+    }
+  }
+
   @VisibleForTesting
   public void setReplicaDeletionDelay(long durationMills) {
     this.replicaDeletionDelay = durationMills;