diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/ContainerSet.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/ContainerSet.java index 13bf7789929d..d122dfb587f9 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/ContainerSet.java +++ b/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> { 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> containerMap = new ConcurrentSkipListMap<>(); private final ConcurrentSkipListSet 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. diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/DeleteBlocksCommandHandler.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/DeleteBlocksCommandHandler.java index 4d5f7931e218..3aa8b10ecc91 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/DeleteBlocksCommandHandler.java +++ b/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) diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/diskbalancer/DiskBalancerService.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/diskbalancer/DiskBalancerService.java index bf22d811ff23..b1bac6771780 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/diskbalancer/DiskBalancerService.java +++ b/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; diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java index 6f93bcef4fd4..474b74586448 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java @@ -1511,10 +1511,17 @@ private ContainerProtos.ContainerChecksumInfo updateAndGetContainerChecksum(Cont @Override public void markContainerUnhealthy(Container container, ScanResult reason) throws IOException { - container.writeLock(); long containerID = container.getContainerData().getContainerID(); + Container lockedContainer = containerSet.getContainerWithWriteLock(containerID); + if (lockedContainer == null) { + // null means container retries exhausted ; + // container not-found throws StorageContainerException. + LOG.warn("Exceeded {} attempts locking live container {}; skipping markContainerUnhealthy.", + ContainerSet.maxContainerMapSwapRetries(), containerID); + return; + } try { - if (container.getContainerState() == State.UNHEALTHY) { + if (lockedContainer.getContainerState() == State.UNHEALTHY) { LOG.debug("Call to mark already unhealthy container {} as unhealthy", containerID); return; @@ -1522,25 +1529,25 @@ public void markContainerUnhealthy(Container container, ScanResult reason) // If the volume is unhealthy, no action is needed. The container has // already been discarded and SCM notified. Once a volume is failed, it // cannot be restored without a restart. - HddsVolume containerVolume = container.getContainerData().getVolume(); + HddsVolume containerVolume = lockedContainer.getContainerData().getVolume(); if (containerVolume.isFailed()) { LOG.debug("Ignoring unhealthy container {} detected on an " + "already failed volume {}", containerID, containerVolume); return; } - container.markContainerUnhealthy(); + lockedContainer.markContainerUnhealthy(); } catch (StorageContainerException ex) { LOG.warn("Unexpected error while marking container {} unhealthy", containerID, ex); } finally { - container.writeUnlock(); + lockedContainer.writeUnlock(); } - updateContainerChecksumFromMetadataIfNeeded(container); + updateContainerChecksumFromMetadataIfNeeded(lockedContainer); // Even if the container file is corrupted/missing and the unhealthy // update fails, the unhealthy state is kept in memory and sent to // SCM. Write a corresponding entry to the container log as well. - ContainerLogger.logUnhealthy(container.getContainerData(), reason); - sendICR(container); + ContainerLogger.logUnhealthy(lockedContainer.getContainerData(), reason); + sendICR(lockedContainer); } @Override @@ -1574,17 +1581,24 @@ public void quasiCloseContainer(Container container, String reason) @Override public void closeContainer(Container container) throws IOException { - container.writeLock(); + long containerID = container.getContainerData().getContainerID(); + Container lockedContainer = containerSet.getContainerWithWriteLock(containerID); + if (lockedContainer == null) { + // null means container locking retries exhausted ; + // container not-found throws StorageContainerException. + LOG.warn("Exceeded {} attempts locking live container {}; skipping closeContainer.", + ContainerSet.maxContainerMapSwapRetries(), containerID); + return; + } try { - final State state = container.getContainerState(); + final State state = lockedContainer.getContainerState(); // Close call is idempotent. if (state == State.CLOSED) { return; } if (state == State.UNHEALTHY) { throw new StorageContainerException( - "Cannot close container #" + container.getContainerData() - .getContainerID() + " while in " + state + " state.", + "Cannot close container #" + containerID + " while in " + state + " state.", ContainerProtos.Result.CONTAINER_UNHEALTHY); } // The container has to be either in CLOSING or in QUASI_CLOSED state. @@ -1593,16 +1607,15 @@ public void closeContainer(Container container) state == State.INVALID ? INVALID_CONTAINER_STATE : CONTAINER_INTERNAL_ERROR; throw new StorageContainerException( - "Cannot close container #" + container.getContainerData() - .getContainerID() + " while in " + state + " state.", error); + "Cannot close container #" + containerID + " while in " + state + " state.", error); } - container.close(); + lockedContainer.close(); } finally { - container.writeUnlock(); + lockedContainer.writeUnlock(); } - updateContainerChecksumFromMetadataIfNeeded(container); - ContainerLogger.logClosed(container.getContainerData()); - sendICR(container); + updateContainerChecksumFromMetadataIfNeeded(lockedContainer); + ContainerLogger.logClosed(lockedContainer.getContainerData()); + sendICR(lockedContainer); } @Override @@ -2293,24 +2306,33 @@ private boolean logBlocksFoundOnDisk(Container container) throws IOException { private void deleteInternal(Container container, boolean force) throws StorageContainerException { + final long containerId = container.getContainerData().getContainerID(); long startTime = clock.millis(); - container.writeLock(); + Container containerLocked = containerSet.getContainerWithWriteLock(containerId); + if (containerLocked == null) { + // null means container locking retries exhausted ; + // container not-found throws StorageContainerException. + LOG.info("Exceeded {} retries to lock container {}; Now SCM will resend for delete with " + + "the current container replica", ContainerSet.maxContainerMapSwapRetries(), + containerId); + return; + } try { - final ContainerData data = container.getContainerData(); - if (container.getContainerData().getVolume().isFailed()) { + final ContainerData data = containerLocked.getContainerData(); + if (containerLocked.getContainerData().getVolume().isFailed()) { // if the volume in which the container resides fails // don't attempt to delete/move it. When a volume fails, // failedVolumeListener will pick it up and clear the container // from the container set. LOG.info("Delete container issued on containerID {} which is in a " + - "failed volume. Skipping", container.getContainerData() + "failed volume. Skipping", containerLocked.getContainerData() .getContainerID()); return; } // If force is false, we check container state. if (!force) { // Check if container is open - if (container.getContainerData().isOpen()) { + if (containerLocked.getContainerData().isOpen()) { throw new StorageContainerException( "Deletion of Open Container is not allowed.", DELETE_ON_OPEN_CONTAINER); @@ -2319,14 +2341,14 @@ private void deleteInternal(Container container, boolean force) // If the container is not empty, it should not be deleted unless the // container is being forcefully deleted (which happens when // container is unhealthy or over-replicated). - if (container.hasBlocks()) { + if (containerLocked.hasBlocks()) { metrics.incContainerDeleteFailedNonEmpty(); LOG.error("Received container deletion command for non-empty {}: {}", data, data.getStatistics()); // blocks table for future debugging. // List blocks - logBlocksIfNonZero(container); + logBlocksIfNonZero(containerLocked); // Log chunks - logBlocksFoundOnDisk(container); + logBlocksFoundOnDisk(containerLocked); throw new StorageContainerException("Non-force deletion of " + "non-empty container is not allowed.", DELETE_ON_NON_EMPTY_CONTAINER); @@ -2334,9 +2356,9 @@ private void deleteInternal(Container container, boolean force) } else { metrics.incContainersForceDelete(); } - if (container.getContainerData() instanceof KeyValueContainerData) { + if (containerLocked.getContainerData() instanceof KeyValueContainerData) { KeyValueContainerData keyValueContainerData = - (KeyValueContainerData) container.getContainerData(); + (KeyValueContainerData) containerLocked.getContainerData(); HddsVolume hddsVolume = keyValueContainerData.getVolume(); // Steps to delete @@ -2350,21 +2372,20 @@ private void deleteInternal(Container container, boolean force) if (waitTime > maxDeleteLockWaitMs) { LOG.warn("An attempt to delete container {} took {} ms acquiring locks and pre-checks. " + "The delete has been skipped and should be retried automatically by SCM.", - container.getContainerData().getContainerID(), waitTime); + containerLocked.getContainerData().getContainerID(), waitTime); return; } - container.markContainerForDelete(); - long containerId = container.getContainerData().getContainerID(); + containerLocked.markContainerForDelete(); containerSet.removeContainer(containerId); - ContainerLogger.logDeleted(container.getContainerData(), force); + ContainerLogger.logDeleted(containerLocked.getContainerData(), force); KeyValueContainerUtil.removeContainer(keyValueContainerData, conf); } catch (IOException ioe) { LOG.error("Failed to move container under " + hddsVolume .getDeletedContainerDir()); String errorMsg = - "Failed to move container" + container.getContainerData() + "Failed to move container" + containerLocked.getContainerData() .getContainerID(); - triggerVolumeScanAndThrowException(container, errorMsg, + triggerVolumeScanAndThrowException(containerLocked, errorMsg, CONTAINER_INTERNAL_ERROR); } } @@ -2374,20 +2395,20 @@ private void deleteInternal(Container container, boolean force) // All other IO Exceptions should be treated as if the container is not // empty as a defensive check. LOG.error("Could not determine if the container {} is empty", - container.getContainerData().getContainerID(), e); + containerLocked.getContainerData().getContainerID(), e); String errorMsg = - "Failed to read container dir" + container.getContainerData() + "Failed to read container dir" + containerLocked.getContainerData() .getContainerID(); - triggerVolumeScanAndThrowException(container, errorMsg, + triggerVolumeScanAndThrowException(containerLocked, errorMsg, CONTAINER_INTERNAL_ERROR); } finally { - container.writeUnlock(); + containerLocked.writeUnlock(); } // Avoid holding write locks for disk operations - sendICR(container); - long bytesUsed = container.getContainerData().getBytesUsed(); - HddsVolume volume = container.getContainerData().getVolume(); - container.delete(); + sendICR(containerLocked); + long bytesUsed = containerLocked.getContainerData().getBytesUsed(); + HddsVolume volume = containerLocked.getContainerData().getVolume(); + containerLocked.delete(); volume.decrementUsedSpace(bytesUsed); } diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/statemachine/background/BlockDeletingTask.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/statemachine/background/BlockDeletingTask.java index 1f66cad476f7..a4ab1a7ef0e8 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/statemachine/background/BlockDeletingTask.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/statemachine/background/BlockDeletingTask.java @@ -44,6 +44,7 @@ import org.apache.hadoop.ozone.container.common.helpers.BlockData; import org.apache.hadoop.ozone.container.common.helpers.BlockDeletingServiceMetrics; import org.apache.hadoop.ozone.container.common.impl.BlockDeletingService; +import org.apache.hadoop.ozone.container.common.impl.ContainerSet; import org.apache.hadoop.ozone.container.common.interfaces.Container; import org.apache.hadoop.ozone.container.common.interfaces.DBHandle; import org.apache.hadoop.ozone.container.common.interfaces.Handler; @@ -67,7 +68,7 @@ public class BlockDeletingTask implements BackgroundTask { private final BlockDeletingServiceMetrics metrics; private final int priority; - private final KeyValueContainerData containerData; + private KeyValueContainerData containerData; private long blocksToDelete; private final OzoneContainer ozoneContainer; private final ConfigurationSource conf; @@ -139,24 +140,39 @@ public BackgroundTaskResult call() throws Exception { private ContainerBackgroundTaskResult handleDeleteTask() throws Exception { ContainerBackgroundTaskResult crr; - final Container container = ozoneContainer.getContainerSet() - .getContainer(containerData.getContainerID()); - container.writeLock(); - File dataDir = new File(containerData.getChunksPath()); - long startTime = Time.monotonicNow(); - // Scan container's db and get list of under deletion blocks - try (DBHandle meta = BlockUtils.getDB(containerData, conf)) { - if (containerData.hasSchema(SCHEMA_V1)) { - crr = deleteViaSchema1(meta, container, dataDir, startTime); - } else if (containerData.hasSchema(SCHEMA_V2)) { - crr = deleteViaSchema2(meta, container, dataDir, startTime); - } else if (containerData.hasSchema(SCHEMA_V3)) { - crr = deleteViaSchema3(meta, container, dataDir, startTime); - } else { - throw new UnsupportedOperationException( - "Only schema version 1,2,3 are supported."); + ContainerSet cs = ozoneContainer.getContainerSet(); + final long containerId = containerData.getContainerID(); + + Container container = cs.getContainerWithWriteLock(containerId); + if (container == null) { + // null means container locking retries exhausted ; + // container not-found throws StorageContainerException. + LOG.info("Exceeded {} retries to lock container {}; Now DN will resend for delete with " + + "the current container replica", ContainerSet.maxContainerMapSwapRetries(), + containerId); + return new ContainerBackgroundTaskResult(); + } + try { + // Always use ContainerData from the locked live Container so paths / RocksDB locations match deleteViaSchema*. + containerData = (KeyValueContainerData) container.getContainerData(); + + File dataDir = new File(containerData.getChunksPath()); + long startTime = Time.monotonicNow(); + // Scan container's db and get list of under deletion blocks + try (DBHandle meta = BlockUtils.getDB(containerData, conf)) { + if (containerData.hasSchema(SCHEMA_V1)) { + crr = deleteViaSchema1(meta, container, dataDir, startTime); + } else if (containerData.hasSchema(SCHEMA_V2)) { + crr = deleteViaSchema2(meta, container, dataDir, startTime); + } else if (containerData.hasSchema(SCHEMA_V3)) { + crr = deleteViaSchema3(meta, container, dataDir, startTime); + } else { + throw new UnsupportedOperationException( + "Only schema version 1,2,3 are supported."); + } + return crr; + } - return crr; } finally { container.writeUnlock(); } diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/impl/TestContainerSet.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/impl/TestContainerSet.java index efb4be86e8dc..33b87f220ed4 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/impl/TestContainerSet.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/impl/TestContainerSet.java @@ -23,6 +23,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.junit.jupiter.api.Assertions.fail; @@ -30,6 +31,10 @@ import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import java.io.IOException; @@ -55,6 +60,7 @@ import org.apache.hadoop.ozone.container.keyvalue.KeyValueContainer; import org.apache.hadoop.ozone.container.keyvalue.KeyValueContainerData; import org.apache.hadoop.ozone.container.ozoneimpl.OnDemandContainerScanner; +import org.junit.jupiter.api.Test; /** * Class used to test ContainerSet operations. @@ -371,6 +377,98 @@ public void testContainerScanHandlerWithoutGap(ContainerLayoutVersion layout) th assertEquals(1, invocationCount.get()); } + // ------------------------------------------------------------------------- + // getContainerWithWriteLock tests + // ------------------------------------------------------------------------- + + /** + * Happy path: container is in the map and the mapping is stable. + * Expect the locked container to be returned, with writeLock called and writeUnlock NOT yet called + */ + @Test + public void testAcquireContainerLockStableMapping() throws StorageContainerException { + ContainerSet cs = spy(newContainerSet()); + Container c1 = mock(Container.class); + doAnswer(inv -> c1).when(cs).getContainer(1L); + + Container result = cs.getContainerWithWriteLock(1L); + + assertSame(c1, result); + verify(c1).writeLock(); + verify(c1, never()).writeUnlock(); + } + + /** + * Container is present when first fetched, but removed from the map after writeLock is acquired + * (second getContainer check returns null). + * Expect StorageContainerException(CONTAINER_NOT_FOUND), and the lock released before throwing (no lock leak). + */ + @Test + public void testContainerRemovedAfterWriteLock() { + ContainerSet cs = spy(newContainerSet()); + Container c1 = mock(Container.class); + int[] callCount = {0}; + // First call → candidate c1; second call (re-check after lock) → null (container removed) + doAnswer(inv -> callCount[0]++ == 0 ? c1 : null).when(cs).getContainer(1L); + + assertThrows(StorageContainerException.class, () -> cs.getContainerWithWriteLock(1L)); + verify(c1).writeLock(); + verify(c1).writeUnlock(); // lock must be released before throwing + } + + /** + * Mapping is swapped once (DiskBalancer moves container from C1 to C2) while the lock is being + * acquired. The first attempt detects the mismatch (current=C2 ≠ candidate=C1), releases C1's lock, + * and retries. The second attempt finds C2 stable and returns it locked. + */ + @Test + public void testRetriesOnMappingSwapThenSucceeds() + throws StorageContainerException { + ContainerSet cs = spy(newContainerSet()); + Container c1 = mock(Container.class); + Container c2 = mock(Container.class); + // Sequence: c1 (candidate retry-0), c2 (current retry-0 → mismatch), + // c2 (candidate retry-1), c2 (current retry-1 → match) + int[] n = {0}; + Container[] seq = {c1, c2, c2, c2}; + doAnswer(inv -> seq[Math.min(n[0]++, seq.length - 1)]).when(cs).getContainer(1L); + + Container result = cs.getContainerWithWriteLock(1L); + + assertSame(c2, result); + // c1 was locked then released during the retry + verify(c1).writeLock(); + verify(c1).writeUnlock(); + // c2 was locked and is held by the caller + verify(c2).writeLock(); + verify(c2, never()).writeUnlock(); + } + + /** + * The mapping keeps changing on every retry. After {@link ContainerSet#maxContainerMapSwapRetries()} + * retries, null is returned. All intermediate locks on C1 must be released (no lock leak). + */ + @Test + public void testExhaustsMaxRetriesReturnsNull() + throws StorageContainerException { + ContainerSet cs = spy(newContainerSet()); + Container c1 = mock(Container.class); + Container c2 = mock(Container.class); + // Alternate: c1 as candidate, c2 as current → always mismatched → all retries fail + int[] n = {0}; + doAnswer(inv -> n[0]++ % 2 == 0 ? c1 : c2).when(cs).getContainer(1L); + + Container result = cs.getContainerWithWriteLock(1L); + + assertNull(result); + int maxRetries = ContainerSet.maxContainerMapSwapRetries(); + // c1 is locked and released once per retry + verify(c1, times(maxRetries)).writeLock(); + verify(c1, times(maxRetries)).writeUnlock(); + // c2 is only ever seen as "current" — it is never locked + verify(c2, never()).writeLock(); + } + /** * Verify that {@code result} contains {@code count} containers * with IDs in increasing order starting at {@code startId}. diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/TestDeleteBlocksCommandHandler.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/TestDeleteBlocksCommandHandler.java index 2b6b387dbe79..6ac39d9ba4f3 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/TestDeleteBlocksCommandHandler.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/TestDeleteBlocksCommandHandler.java @@ -29,10 +29,12 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.any; import static org.mockito.Mockito.doAnswer; -import static org.mockito.Mockito.eq; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; import static org.mockito.Mockito.spy; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; @@ -53,6 +55,7 @@ import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import org.apache.hadoop.hdds.HddsConfigKeys; import org.apache.hadoop.hdds.conf.OzoneConfiguration; import org.apache.hadoop.hdds.protocol.DatanodeDetails; @@ -96,6 +99,7 @@ public class TestDeleteBlocksCommandHandler { private String schemaVersion; private HddsVolume volume1; private BlockDeletingServiceMetrics blockDeleteMetrics; + private static final long STALE_CONTAINER_ID = 5L; private void prepareTest(ContainerTestVersionInfo versionInfo) throws Exception { @@ -481,6 +485,89 @@ public void testDuplicateTxFromSCMHandledByDeleteBlocksCommandHandler( assertEquals(afterSecondPendingBytes + 768L, containerData.getBlockPendingDeletionBytes()); } + /** + * Simulates DiskBalancer swapping the {@link ContainerSet} entry after the + * delete-blocks worker read the container once but before it re-checks after + * {@code writeLockTryLock}: the first attempt must treat {@code containerData} as stale, fail + * without calling the schema handler on the old replica, and succeed on the built-in retry + * against the live replica. + */ + @Test + public void deleteBlocksRetriesWhenContainerDataStale() throws Exception { + schemaVersion = SCHEMA_V3; + OzoneConfiguration conf = new OzoneConfiguration(); + ContainerTestVersionInfo.setTestSchemaVersion(schemaVersion, conf); + OzoneContainer ozoneContainer = mock(OzoneContainer.class); + ContainerLayoutVersion layout = ContainerLayoutVersion.FILE_PER_BLOCK; + ContainerSet realSet = newContainerSet(); + volume1 = mockHddsVolume("uuid-1"); + for (int i = 0; i <= 10; i++) { + KeyValueContainerData data = + new KeyValueContainerData(i, + layout, + ContainerTestHelper.CONTAINER_MAX_SIZE, + UUID.randomUUID().toString(), + UUID.randomUUID().toString()); + data.setSchemaVersion(schemaVersion); + data.setVolume(volume1); + KeyValueContainer container = new KeyValueContainer(data, conf); + data.closeContainer(); + realSet.addContainer(container); + } + + KeyValueContainer oldContainer = + (KeyValueContainer) realSet.getContainer(STALE_CONTAINER_ID); + KeyValueContainerData newReplicaData = + new KeyValueContainerData((KeyValueContainerData) oldContainer.getContainerData()); + newReplicaData.setVolume(volume1); + newReplicaData.closeContainer(); + KeyValueContainer newContainer = new KeyValueContainer(newReplicaData, conf); + + AtomicInteger getContainerSequence = new AtomicInteger(0); + ContainerSet containerSetSpy = spy(realSet); + doAnswer(invocation -> { + long id = invocation.getArgument(0); + if (id != STALE_CONTAINER_ID) { + return invocation.callRealMethod(); + } + int seq = getContainerSequence.getAndIncrement(); + if (seq == 0) { + return oldContainer; + } + if (seq == 1) { + return newContainer; + } + return newContainer; + }).when(containerSetSpy).getContainer(anyLong()); + + when(ozoneContainer.getContainerSet()).thenReturn(containerSetSpy); + containerSet = containerSetSpy; + + DatanodeConfiguration dnConf = conf.getObject(DatanodeConfiguration.class); + handler = spy(new DeleteBlocksCommandHandler(ozoneContainer, conf, dnConf, "")); + blockDeleteMetrics = handler.getBlockDeleteMetrics(); + TestSchemaHandler testSchemaHandler1 = spy(new TestSchemaHandler()); + TestSchemaHandler testSchemaHandler2 = spy(new TestSchemaHandler()); + TestSchemaHandler testSchemaHandler3 = spy(new TestSchemaHandler()); + handler.getSchemaHandlers().put(SCHEMA_V1, testSchemaHandler1); + handler.getSchemaHandlers().put(SCHEMA_V2, testSchemaHandler2); + handler.getSchemaHandlers().put(SCHEMA_V3, testSchemaHandler3); + + DeletedBlocksTransaction transaction = + createDeletedBlocksTransaction(77L, STALE_CONTAINER_ID); + List results = + handler.executeCmdWithRetry(Collections.singletonList(transaction)); + + assertEquals(1, results.size()); + assertTrue(results.get(0).getSuccess()); + verify(handler, times(2)).submitTasks(any()); + verify(handler.getSchemaHandlers().get(SCHEMA_V3), times(1)) + .handle(eq(newReplicaData), eq(transaction)); + verify(handler.getSchemaHandlers().get(SCHEMA_V3), never()) + .handle(eq((KeyValueContainerData) oldContainer.getContainerData()), any()); + assertEquals(0, blockDeleteMetrics.getTotalLockTimeoutTransactionCount()); + } + private DeletedBlocksTransaction createDeletedBlocksTransaction(long txID, long containerID) { return DeletedBlocksTransaction.newBuilder() diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/diskbalancer/TestDiskBalancerWithConcurrentBackgroundTasks.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/diskbalancer/TestDiskBalancerWithConcurrentBackgroundTasks.java new file mode 100644 index 000000000000..2463d0f872fd --- /dev/null +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/diskbalancer/TestDiskBalancerWithConcurrentBackgroundTasks.java @@ -0,0 +1,591 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hadoop.ozone.container.diskbalancer; + +import static org.apache.hadoop.ozone.container.common.ContainerTestUtils.createDbInstancesForTestIfNeeded; +import static org.apache.hadoop.ozone.container.common.ContainerTestUtils.getUnhealthyDataScanResult; +import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.io.File; +import java.io.IOException; +import java.util.Arrays; +import java.util.Comparator; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import org.apache.commons.io.FileUtils; +import org.apache.hadoop.hdds.HddsConfigKeys; +import org.apache.hadoop.hdds.conf.OzoneConfiguration; +import org.apache.hadoop.hdds.fs.MockSpaceUsageCheckFactory; +import org.apache.hadoop.hdds.fs.SpaceUsageCheckFactory; +import org.apache.hadoop.hdds.fs.SpaceUsageSource; +import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos; +import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerDataProto.State; +import org.apache.hadoop.hdds.scm.ScmConfigKeys; +import org.apache.hadoop.hdds.utils.FaultInjector; +import org.apache.hadoop.hdds.utils.db.Table; +import org.apache.hadoop.ozone.container.checksum.ContainerChecksumTreeManager; +import org.apache.hadoop.ozone.container.common.helpers.ContainerMetrics; +import org.apache.hadoop.ozone.container.common.helpers.ContainerUtils; +import org.apache.hadoop.ozone.container.common.impl.BlockDeletingService; +import org.apache.hadoop.ozone.container.common.impl.ContainerDataYaml; +import org.apache.hadoop.ozone.container.common.impl.ContainerSet; +import org.apache.hadoop.ozone.container.common.interfaces.Container; +import org.apache.hadoop.ozone.container.common.interfaces.ContainerDispatcher; +import org.apache.hadoop.ozone.container.common.interfaces.DBHandle; +import org.apache.hadoop.ozone.container.common.interfaces.Handler; +import org.apache.hadoop.ozone.container.common.interfaces.ScanResult; +import org.apache.hadoop.ozone.container.common.interfaces.VolumeChoosingPolicy; +import org.apache.hadoop.ozone.container.common.volume.HddsVolume; +import org.apache.hadoop.ozone.container.common.volume.MutableVolumeSet; +import org.apache.hadoop.ozone.container.common.volume.StorageVolume; +import org.apache.hadoop.ozone.container.common.volume.VolumeSet; +import org.apache.hadoop.ozone.container.keyvalue.ContainerTestVersionInfo; +import org.apache.hadoop.ozone.container.keyvalue.KeyValueContainer; +import org.apache.hadoop.ozone.container.keyvalue.KeyValueContainerData; +import org.apache.hadoop.ozone.container.keyvalue.KeyValueHandler; +import org.apache.hadoop.ozone.container.keyvalue.helpers.BlockUtils; +import org.apache.hadoop.ozone.container.keyvalue.statemachine.background.BlockDeletingTask; +import org.apache.hadoop.ozone.container.ozoneimpl.ContainerController; +import org.apache.hadoop.ozone.container.ozoneimpl.OzoneContainer; +import org.apache.ozone.test.GenericTestUtils; +import org.apache.ozone.test.GenericTestUtils.LogCapturer; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.api.io.TempDir; + +/** + * The balancer thread holds a read lock on the old replica while it copies the + * container and then calls {@link ContainerSet#updateContainer} so the map points at the new + * replica on another disk. Concurrent delete / block-deletion / unhealthy / close paths resolve + * the live container by id via {@link ContainerSet#getContainerWithWriteLock} and then operate on + * that instance so paths, DB, and state match the destination replica. + */ +@Timeout(60) +class TestDiskBalancerWithConcurrentBackgroundTasks { + + @TempDir + private java.nio.file.Path tmpDir; + + private File testDir; + private final String scmId = UUID.randomUUID().toString(); + private final String datanodeUuid = UUID.randomUUID().toString(); + private final OzoneConfiguration conf = new OzoneConfiguration(); + + private OzoneContainer ozoneContainer; + private ContainerSet containerSet; + private MutableVolumeSet volumeSet; + private KeyValueHandler keyValueHandler; + private ContainerChecksumTreeManager checksumTreeManager; + private DiskBalancerServiceTestImpl diskBalancerService; + + private HddsVolume hotVolume; + private HddsVolume coldVolume; + + private static final long CONTAINER_ID = 42L; + private static final long CONTAINER_SIZE = 1024L * 1024L; + private static final long SEEDED_PENDING_BLOCKS = 2L; + private static final long SEEDED_PENDING_BYTES = 2048L; + + /** + * Pauses disk balancer immediately after {@link ContainerSet#updateContainer} (map points at + * the new replica) but before readUnlock, so another thread can run while the balancer + * still holds readLock on the old container. + */ + private static final class AfterInMemoryUpdateInjector extends FaultInjector { + private final CountDownLatch reachedSwapPoint = new CountDownLatch(1); + private final CountDownLatch continueBalancer = new CountDownLatch(1); + + @Override + public void pause() throws IOException { + // Signal the test thread that the race window has started, then block the balancer. + reachedSwapPoint.countDown(); + try { + continueBalancer.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException(e); + } + } + + void awaitSwapPoint() throws InterruptedException, TimeoutException { + if (!reachedSwapPoint.await(60, TimeUnit.SECONDS)) { + throw new TimeoutException("balancer did not reach post-updateContainer hook"); + } + } + + // Unblock the balancer so it can readUnlock and finish the move. + void continueBalancer() { + continueBalancer.countDown(); + } + } + + @BeforeEach + void setup() throws Exception { + testDir = tmpDir.toFile(); + conf.set(HddsConfigKeys.OZONE_METADATA_DIRS, testDir.getAbsolutePath()); + + conf.set(ScmConfigKeys.HDDS_DATANODE_DIR_KEY, + testDir.getAbsolutePath() + "/vol0," + + testDir.getAbsolutePath() + "/vol1," + + testDir.getAbsolutePath() + "/vol2"); + conf.setClass(SpaceUsageCheckFactory.Conf.configKeyForClassName(), + MockSpaceUsageCheckFactory.HalfTera.class, + SpaceUsageCheckFactory.class); + + volumeSet = new MutableVolumeSet(datanodeUuid, scmId, conf, null, + StorageVolume.VolumeType.DATA_VOLUME, null); + createDbInstancesForTestIfNeeded(volumeSet, scmId, scmId, conf); + + List volumes = volumeSet.getVolumesList(); + HddsVolume v0 = (HddsVolume) volumes.get(0); + HddsVolume v1 = (HddsVolume) volumes.get(1); + HddsVolume v2 = (HddsVolume) volumes.get(2); + + for (HddsVolume v : new HddsVolume[] {v0, v1, v2}) { + v.incrementUsedSpace(0 - v.getCurrentUsage().getUsedSpace()); + } + + long capacity = v0.getCurrentUsage().getCapacity(); + v0.incrementUsedSpace(capacity / 20); + v1.incrementUsedSpace(capacity / 20); + v2.incrementUsedSpace(capacity / 2); + + coldVolume = coldestVolume(v0, v1, v2); + hotVolume = hottestVolume(v0, v1, v2); + + containerSet = ContainerSet.newReadOnlyContainerSet(1000); + ContainerMetrics containerMetrics = ContainerMetrics.create(conf); + checksumTreeManager = new ContainerChecksumTreeManager(conf); + keyValueHandler = new KeyValueHandler(conf, datanodeUuid, + containerSet, volumeSet, containerMetrics, c -> { }, + checksumTreeManager); + keyValueHandler.setClusterID(scmId); + + Map handlers = new HashMap<>(); + handlers.put(ContainerProtos.ContainerType.KeyValueContainer, keyValueHandler); + ContainerController controller = new ContainerController(containerSet, handlers); + ContainerDispatcher dispatcher = mock(ContainerDispatcher.class); + when(dispatcher.getHandler(ContainerProtos.ContainerType.KeyValueContainer)) + .thenReturn(keyValueHandler); + + ozoneContainer = mock(OzoneContainer.class); + when(ozoneContainer.getContainerSet()).thenReturn(containerSet); + when(ozoneContainer.getVolumeSet()).thenReturn(volumeSet); + when(ozoneContainer.getController()).thenReturn(controller); + when(ozoneContainer.getDispatcher()).thenReturn(dispatcher); + + DiskBalancerConfiguration diskBalancerConfiguration = conf.getObject(DiskBalancerConfiguration.class); + diskBalancerConfiguration.setDiskBalancerShouldRun(true); + conf.setFromObject(diskBalancerConfiguration); + diskBalancerService = new DiskBalancerServiceTestImpl(ozoneContainer, 100, conf, 1); + // Immediate cleanup of the source replica after a move + diskBalancerService.setReplicaDeletionDelay(0); + } + + @AfterEach + void cleanup() throws IOException { + DiskBalancerService.setInjector(null); + if (diskBalancerService != null) { + diskBalancerService.shutdown(); + } + BlockUtils.shutdownCache(conf); + if (volumeSet != null) { + volumeSet.shutdown(); + } + if (testDir != null && testDir.exists()) { + FileUtils.deleteDirectory(testDir); + } + } + + /** + * Force-delete with a stale {@link Container} handle still targets the container id; after a + * DiskBalancer swap, {@code deleteInternal} locks the live map entry (destination replica) and + * applies deletion there — not on the old source object passed in from the RPC. + */ + @ContainerTestVersionInfo.ContainerTest + void containerDeleteStaleRefKeepsSwappedReplica(ContainerTestVersionInfo versionInfo) + throws Exception { + // Capture delete path logs: deleteInternal and diskBalancer + LogCapturer kvLogs = LogCapturer.captureLogs(KeyValueHandler.class); + LogCapturer diskBalancerLogs = LogCapturer.captureLogs(DiskBalancerService.class); + + String schemaVersion = versionInfo.getSchemaVersion(); + ContainerTestVersionInfo.setTestSchemaVersion(schemaVersion, conf); + + KeyValueContainer oldReplica = createClosedContainer(CONTAINER_ID, hotVolume, versionInfo); + Container staleContainerRef = oldReplica; + String oldReplicaPathOnHot = oldReplica.getContainerData().getContainerPath(); + assertThat(new File(oldReplicaPathOnHot)).exists(); + + // Install injector so balancer pauses right after ContainerSet points at the new location of replica. + AfterInMemoryUpdateInjector raceInjector = new AfterInMemoryUpdateInjector(); + DiskBalancerService.setInjector(raceInjector); + + DiskBalancerService.DiskBalancerTask task = + (DiskBalancerService.DiskBalancerTask) diskBalancerService.getTasks().poll(); + assertNotNull(task); + + // Run the move on a background thread; it will block inside the injector. + CompletableFuture balancerDone = + CompletableFuture.runAsync(() -> task.call()); + + // Wait until updateContainer(newReplica) is done; balancer still holds readLock on old replica. + raceInjector.awaitSwapPoint(); + + Container currentContainerRef = containerSet.getContainer(CONTAINER_ID); + assertNotNull(currentContainerRef); + assertNotEquals(staleContainerRef, currentContainerRef, + "ContainerSet should reference the new replica before readUnlock"); + assertEquals(coldVolume, currentContainerRef.getContainerData().getVolume(), + "Replica should already be on the destination volume"); + + String destinationPathBeforeDelete = + currentContainerRef.getContainerData().getContainerPath(); + + // Start RM-style force delete using the stale Container handle; deleteInternal resolves id 42, + // acquires writeLock on the live map entry (destination replica), not on the stale source handle. + CountDownLatch deleteThreadPastSchedule = new CountDownLatch(1); + CompletableFuture deleteDone = CompletableFuture.runAsync(() -> { + deleteThreadPastSchedule.countDown(); + try { + keyValueHandler.deleteContainer(staleContainerRef, true); + } catch (IOException e) { + throw new RuntimeException(e); + } + }); + assertThat(deleteThreadPastSchedule.await(10, TimeUnit.SECONDS)).isTrue(); + Thread.sleep(200); + + raceInjector.continueBalancer(); + + CompletableFuture.allOf(balancerDone, deleteDone).get(60, TimeUnit.SECONDS); + + assertThat(kvLogs.getOutput()).doesNotContain("reference is stale"); + + // Old replica: disk balancer marks it DELETED after the move. + assertEquals(State.DELETED, staleContainerRef.getContainerState()); + assertEquals(hotVolume, staleContainerRef.getContainerData().getVolume()); + + // Live map entry was removed by force-delete on the destination replica. + assertNull(containerSet.getContainer(CONTAINER_ID)); + assertThat(new File(destinationPathBeforeDelete)).doesNotExist(); + + // Disk balancer delayed cleanup removes the old replica from the source path — not RM delete. + assertThat(new File(oldReplicaPathOnHot)).doesNotExist(); + GenericTestUtils.waitFor( + () -> diskBalancerLogs.getOutput().contains("Deleted expired container 42 after delay") + && diskBalancerLogs.getOutput().contains(String.valueOf(CONTAINER_ID)), + 100, 10_000); + } + + /** + * BlockDeletingTask queued with stale ref of KeyValueContainerData still resolves + * the live replica by id; after {@code getContainerWithWriteLock}, it uses the destination + * replica's DB and paths (updated {@code containerData} field). + */ + @ContainerTestVersionInfo.ContainerTest + void blockTaskStaleDataKeepsPendingOnDestination(ContainerTestVersionInfo versionInfo) + throws Exception { + LogCapturer logs = LogCapturer.captureLogs(BlockDeletingTask.class); + LogCapturer diskBalancerLogs = LogCapturer.captureLogs(DiskBalancerService.class); + String schemaVersion = versionInfo.getSchemaVersion(); + ContainerTestVersionInfo.setTestSchemaVersion(schemaVersion, conf); + + KeyValueContainer oldReplica = createClosedContainer(CONTAINER_ID, hotVolume, versionInfo); + seedPendingDeletionInMetadata(oldReplica); + KeyValueContainerData staleReplicaData = oldReplica.getContainerData(); + + // Balancer injector — pause after map swap, before readUnlock. + AfterInMemoryUpdateInjector raceInjector = new AfterInMemoryUpdateInjector(); + DiskBalancerService.setInjector(raceInjector); + + DiskBalancerService.DiskBalancerTask balancerTask = + (DiskBalancerService.DiskBalancerTask) diskBalancerService.getTasks().poll(); + assertNotNull(balancerTask); + CompletableFuture balancerDone = + CompletableFuture.runAsync(() -> balancerTask.call()); + + // We are past updateContainer; new replica is in ContainerSet; balancer still read-locks old replica. + raceInjector.awaitSwapPoint(); + + // New ContainerData instance must differ from what the queued block task still holds. + Container newReplicaData = containerSet.getContainer(CONTAINER_ID); + assertNotNull(newReplicaData); + assertNotEquals(staleReplicaData, newReplicaData.getContainerData()); + KeyValueContainerData newData = (KeyValueContainerData) newReplicaData.getContainerData(); + long pendingBefore = readPendingDeleteBlockCount(newData); + assertEquals(pendingBefore, SEEDED_PENDING_BLOCKS, + "new replica should report same pending deletions from copied metadata"); + + BlockDeletingService blockDeletingService = + new BlockDeletingService(ozoneContainer, 500, 500, TimeUnit.MILLISECONDS, 1, conf, + checksumTreeManager); + BlockDeletingService.ContainerBlockInfo blockInfo = + new BlockDeletingService.ContainerBlockInfo(staleReplicaData, SEEDED_PENDING_BLOCKS + 10); + BlockDeletingTask blockDeletingTask = + new BlockDeletingTask(blockDeletingService, blockInfo, checksumTreeManager, 1); + + // Run block deletion concurrently; getContainerWithWriteLock targets the live map entry on the destination. + CountDownLatch blockThreadStarted = new CountDownLatch(1); + CompletableFuture blockDone = CompletableFuture.runAsync(() -> { + blockThreadStarted.countDown(); + try { + blockDeletingTask.call(); + } catch (Exception e) { + throw new RuntimeException(e); + } + }); + assertThat(blockThreadStarted.await(10, TimeUnit.SECONDS)).isTrue(); + Thread.sleep(200); + + raceInjector.continueBalancer(); + CompletableFuture.allOf(balancerDone, blockDone).get(60, TimeUnit.SECONDS); + + assertThat(logs.getOutput()).doesNotContain("reference is stale"); + + KeyValueContainerData newContainerData = + (KeyValueContainerData) containerSet.getContainer(CONTAINER_ID).getContainerData(); + assertNotNull(newContainerData); + assertThat(readPendingDeleteBlockCount(newContainerData)).isLessThanOrEqualTo(pendingBefore); + assertThat(new File(newContainerData.getChunksPath())).exists(); + // Disk balancer delayed cleanup removes the old replica from the source path — not RM delete. + assertThat(new File(staleReplicaData.getContainerPath())).doesNotExist(); + GenericTestUtils.waitFor( + () -> diskBalancerLogs.getOutput().contains("Deleted expired container 42 after delay") + && diskBalancerLogs.getOutput().contains(String.valueOf(CONTAINER_ID)), + 100, 10_000); + } + + /** + * {@link KeyValueHandler#markContainerUnhealthy} with a stale container + * reference while DiskBalancer has already run {@link ContainerSet#updateContainer} + * (map = destination). Without {@link ContainerSet#getContainerWithWriteLock}, the handler would + * take writeLock on the old object and mark it unhealthy after {@code markContainerForDelete} + * turns it {@link State#DELETED}, sending a false UNHEALTHY ICR for a replica SCM no longer tracks. + * With the fix, the live map entry (destination) is locked and marked {@link State#UNHEALTHY}. + */ + @ContainerTestVersionInfo.ContainerTest + void markUnhealthyAppliedOnDestVolumeContainer( + ContainerTestVersionInfo versionInfo) throws Exception { + LogCapturer diskBalancerLogs = LogCapturer.captureLogs(DiskBalancerService.class); + String schemaVersion = versionInfo.getSchemaVersion(); + ContainerTestVersionInfo.setTestSchemaVersion(schemaVersion, conf); + + KeyValueContainer oldReplica = createClosedContainer(CONTAINER_ID, hotVolume, versionInfo); + Container staleContainerRef = oldReplica; + + AfterInMemoryUpdateInjector raceInjector = new AfterInMemoryUpdateInjector(); + DiskBalancerService.setInjector(raceInjector); + + DiskBalancerService.DiskBalancerTask balancerTask = + (DiskBalancerService.DiskBalancerTask) diskBalancerService.getTasks().poll(); + assertNotNull(balancerTask); + CompletableFuture balancerDone = + CompletableFuture.runAsync(balancerTask::call); + + raceInjector.awaitSwapPoint(); + + Container liveReplica = containerSet.getContainer(CONTAINER_ID); + assertNotNull(liveReplica); + assertNotEquals(staleContainerRef, liveReplica, + "ContainerSet should reference the destination replica at the race hook"); + assertEquals(State.CLOSED, liveReplica.getContainerState()); + + ScanResult reason = getUnhealthyDataScanResult(); + CountDownLatch unhealthyThreadStarted = new CountDownLatch(1); + CompletableFuture unhealthyDone = CompletableFuture.runAsync(() -> { + unhealthyThreadStarted.countDown(); + try { + keyValueHandler.markContainerUnhealthy(staleContainerRef, reason); + } catch (IOException e) { + throw new RuntimeException(e); + } + }); + assertThat(unhealthyThreadStarted.await(10, TimeUnit.SECONDS)).isTrue(); + Thread.sleep(200); + + raceInjector.continueBalancer(); + CompletableFuture.allOf(balancerDone, unhealthyDone).get(60, TimeUnit.SECONDS); + + Container afterMove = containerSet.getContainer(CONTAINER_ID); + assertNotNull(afterMove); + assertSame(liveReplica, afterMove); + assertEquals(State.UNHEALTHY, afterMove.getContainerState(), + "UNHEALTHY must apply to the live destination replica, not the stale source handle"); + + assertEquals(State.DELETED, staleContainerRef.getContainerState()); + assertEquals(hotVolume, staleContainerRef.getContainerData().getVolume()); + + GenericTestUtils.waitFor( + () -> diskBalancerLogs.getOutput().contains("Deleted expired container 42 after delay") + && diskBalancerLogs.getOutput().contains(String.valueOf(CONTAINER_ID)), + 100, 10_000); + } + + /** + * SCM closeContainer with a stale source container while the map already references + * the destination after {@link ContainerSet#updateContainer}. Without resolving the live instance, + * the close would run on the source after it is DELETED and throw, leaving the destination + * {@link State#QUASI_CLOSED}. With {@link ContainerSet#getContainerWithWriteLock}, the destination + * is closed to {@link State#CLOSED}. + */ + @ContainerTestVersionInfo.ContainerTest + void closeContainerAppliesOnDestVolumeContainer( + ContainerTestVersionInfo versionInfo) throws Exception { + LogCapturer diskBalancerLogs = LogCapturer.captureLogs(DiskBalancerService.class); + String schemaVersion = versionInfo.getSchemaVersion(); + ContainerTestVersionInfo.setTestSchemaVersion(schemaVersion, conf); + + KeyValueContainer oldReplica = createClosedContainer(CONTAINER_ID, hotVolume, versionInfo); + persistQuasiClosedState(oldReplica); + Container staleContainerRef = oldReplica; + + AfterInMemoryUpdateInjector raceInjector = new AfterInMemoryUpdateInjector(); + DiskBalancerService.setInjector(raceInjector); + + DiskBalancerService.DiskBalancerTask balancerTask = + (DiskBalancerService.DiskBalancerTask) diskBalancerService.getTasks().poll(); + assertNotNull(balancerTask); + CompletableFuture balancerDone = + CompletableFuture.runAsync(balancerTask::call); + + raceInjector.awaitSwapPoint(); + + Container liveReplica = containerSet.getContainer(CONTAINER_ID); + assertNotNull(liveReplica); + assertNotEquals(staleContainerRef, liveReplica); + assertEquals(State.QUASI_CLOSED, liveReplica.getContainerState()); + + CountDownLatch closeThreadStarted = new CountDownLatch(1); + CompletableFuture closeDone = CompletableFuture.runAsync(() -> { + closeThreadStarted.countDown(); + try { + keyValueHandler.closeContainer(staleContainerRef); + } catch (IOException e) { + throw new RuntimeException(e); + } + }); + assertThat(closeThreadStarted.await(10, TimeUnit.SECONDS)).isTrue(); + Thread.sleep(200); + + raceInjector.continueBalancer(); + CompletableFuture.allOf(balancerDone, closeDone).get(60, TimeUnit.SECONDS); + + Container afterMove = containerSet.getContainer(CONTAINER_ID); + assertNotNull(afterMove); + assertSame(liveReplica, afterMove); + assertEquals(State.CLOSED, afterMove.getContainerState(), + "CLOSED transition must apply to the live destination replica"); + + assertEquals(State.DELETED, staleContainerRef.getContainerState()); + + GenericTestUtils.waitFor( + () -> diskBalancerLogs.getOutput().contains("Deleted expired container 42 after delay") + && diskBalancerLogs.getOutput().contains(String.valueOf(CONTAINER_ID)), + 100, 10_000); + } + + /** + * Makes {@link State#QUASI_CLOSED} visible on disk so import/copy sees the same state + * the in-memory container had before the move. + */ + private void persistQuasiClosedState(KeyValueContainer container) throws IOException { + KeyValueContainerData data = container.getContainerData(); + data.setState(State.QUASI_CLOSED); + File containerFile = ContainerUtils.getContainerFile(new File(data.getContainerPath())); + ContainerDataYaml.createContainerFile(data, containerFile); + } + + private long readPendingDeleteBlockCount(KeyValueContainerData data) throws IOException { + try (DBHandle db = BlockUtils.getDB(data, conf)) { + Table meta = db.getStore().getMetadataTable(); + Long v = meta.get(data.getPendingDeleteBlockCountKey()); + return v == null ? 0L : v; + } + } + + /** + * Persists pending-deletion counters in container metadata so they survive + * disk balancer copy/import and can be verified on the destination replica. + */ + private void seedPendingDeletionInMetadata(KeyValueContainer container) throws IOException { + KeyValueContainerData data = container.getContainerData(); + try (DBHandle metadata = BlockUtils.getDB(data, conf)) { + Table meta = metadata.getStore().getMetadataTable(); + meta.put(data.getPendingDeleteBlockCountKey(), SEEDED_PENDING_BLOCKS); + meta.put(data.getPendingDeleteBlockBytesKey(), SEEDED_PENDING_BYTES); + } + data.incrPendingDeletionBlocks(SEEDED_PENDING_BLOCKS, SEEDED_PENDING_BYTES); + } + + private static HddsVolume coldestVolume(HddsVolume... volumes) { + return Arrays.stream(volumes) + .min(volumePolicyOrder()) + .get(); + } + + private static HddsVolume hottestVolume(HddsVolume... volumes) { + return Arrays.stream(volumes) + .max(volumePolicyOrder()) + .get(); + } + + private static Comparator volumePolicyOrder() { + return Comparator + .comparingDouble((HddsVolume v) -> { + SpaceUsageSource usage = v.getCurrentUsage(); + return (double) (usage.getCapacity() - usage.getAvailable()) / usage.getCapacity(); + }) + .thenComparing(HddsVolume::getStorageID); + } + + private KeyValueContainer createClosedContainer(long containerId, HddsVolume vol, + ContainerTestVersionInfo versionInfo) + throws IOException { + KeyValueContainerData containerData = new KeyValueContainerData( + containerId, versionInfo.getLayout(), CONTAINER_SIZE, + UUID.randomUUID().toString(), datanodeUuid); + containerData.setState(State.CLOSED); + containerData.getStatistics().setBlockBytesForTesting(CONTAINER_SIZE); + containerData.setSchemaVersion(versionInfo.getSchemaVersion()); + + KeyValueContainer container = new KeyValueContainer(containerData, conf); + VolumeChoosingPolicy policy = mock(VolumeChoosingPolicy.class); + when(policy.chooseVolume(any(List.class), anyLong())).thenReturn(vol); + container.create((VolumeSet) volumeSet, policy, scmId); + containerSet.addContainer(container); + vol.incrementUsedSpace(containerData.getBytesUsed()); + return container; + } +} diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueHandler.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueHandler.java index 0385564ebf75..7ba862389d10 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueHandler.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueHandler.java @@ -370,18 +370,20 @@ private ContainerCommandRequestProto getDummyCommandRequestProto( @ContainerLayoutTestInfo.ContainerTest public void testCloseInvalidContainer(ContainerLayoutVersion layoutVersion) throws IOException { - KeyValueHandler keyValueHandler = createKeyValueHandler(tempDir); conf = new OzoneConfiguration(); - KeyValueContainerData kvData = new KeyValueContainerData(DUMMY_CONTAINER_ID, - layoutVersion, - (long) StorageUnit.GB.toBytes(1), UUID.randomUUID().toString(), - UUID.randomUUID().toString()); - kvData.setMetadataPath(tempDir.toString()); - kvData.setDbFile(dbFile.toFile()); - KeyValueContainer container = new KeyValueContainer(kvData, conf); + conf.set(OZONE_SCM_CONTAINER_LAYOUT_KEY, layoutVersion.name()); + HandlerWithVolumeSet handlerCtx = createKeyValueHandler(tempDir); + KeyValueHandler keyValueHandler = handlerCtx.getHandler(); + ContainerSet containerSet = handlerCtx.getContainerSet(); + + long containerId = DUMMY_CONTAINER_ID + layoutVersion.getVersion(); ContainerCommandRequestProto createContainerRequest = - createContainerRequest(DATANODE_UUID, DUMMY_CONTAINER_ID); - keyValueHandler.handleCreateContainer(createContainerRequest, container); + createContainerRequest(DATANODE_UUID, containerId); + keyValueHandler.handleCreateContainer(createContainerRequest, null); + + KeyValueContainer container = + (KeyValueContainer) containerSet.getContainer(containerId); + KeyValueContainerData kvData = container.getContainerData(); // Make the container state as invalid. kvData.setState(ContainerProtos.ContainerDataProto.State.INVALID); @@ -390,12 +392,11 @@ public void testCloseInvalidContainer(ContainerLayoutVersion layoutVersion) ContainerCommandRequestProto closeContainerRequest = ContainerProtos.ContainerCommandRequestProto.newBuilder() .setCmdType(ContainerProtos.Type.CloseContainer) - .setContainerID(DUMMY_CONTAINER_ID) + .setContainerID(containerId) .setDatanodeUuid(DATANODE_UUID) .setCloseContainer(ContainerProtos.CloseContainerRequestProto .getDefaultInstance()) .build(); - dispatcher.dispatch(closeContainerRequest, null); // Closing invalid container should return error response. ContainerProtos.ContainerCommandResponseProto response = @@ -923,7 +924,7 @@ private static ContainerCommandRequestProto createContainerRequest( .build(); } - private KeyValueHandler createKeyValueHandler(Path path) throws IOException { + private HandlerWithVolumeSet createKeyValueHandler(Path path) throws IOException { final ContainerSet containerSet = newContainerSet(); final MutableVolumeSet volumeSet = mock(MutableVolumeSet.class); @@ -950,7 +951,7 @@ private KeyValueHandler createKeyValueHandler(Path path) throws IOException { conf.getObject(ContainerScannerConfiguration.class), controller); containerSet.registerOnDemandScanner(onDemandScanner); - return kvHandler; + return new HandlerWithVolumeSet(kvHandler, volumeSet, containerSet); } private static class HandlerWithVolumeSet { diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueHandlerWithUnhealthyContainer.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueHandlerWithUnhealthyContainer.java index 8361959e6da4..358b56157fc8 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueHandlerWithUnhealthyContainer.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueHandlerWithUnhealthyContainer.java @@ -29,6 +29,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.any; import static org.mockito.Mockito.atMostOnce; import static org.mockito.Mockito.mock; @@ -220,7 +221,8 @@ void testNPEFromPutBlock() throws IOException { @Test public void testMarkContainerUnhealthyInFailedVolume() throws IOException { - KeyValueHandler handler = getDummyHandler(); + ContainerSet containerSet = mock(ContainerSet.class); + KeyValueHandler handler = getDummyHandler(containerSet); KeyValueContainerData kvData = new KeyValueContainerData(1L, ContainerLayoutVersion.FILE_PER_BLOCK, (long) StorageUnit.GB.toBytes(1), UUID.randomUUID().toString(), @@ -233,6 +235,10 @@ public void testMarkContainerUnhealthyInFailedVolume() throws IOException { .build(); kvData.setVolume(hddsVolume); KeyValueContainer container = new KeyValueContainer(kvData, conf); + when(containerSet.getContainerWithWriteLock(eq(1L))).thenAnswer(invocation -> { + container.writeLock(); + return container; + }); // When volume is failed, the call to mark the container unhealthy should // be ignored. @@ -252,6 +258,10 @@ public void testMarkContainerUnhealthyInFailedVolume() throws IOException { // -- Helper methods below. private KeyValueHandler getDummyHandler() { + return getDummyHandler(mock(ContainerSet.class)); + } + + private KeyValueHandler getDummyHandler(ContainerSet containerSet) { DatanodeDetails dnDetails = DatanodeDetails.newBuilder() .setUuid(UUID.fromString(DATANODE_UUID)) .setHostName("dummyHost") @@ -263,7 +273,7 @@ private KeyValueHandler getDummyHandler() { return new KeyValueHandler( conf, stateMachine.getDatanodeDetails().getUuidString(), - mock(ContainerSet.class), + containerSet, mock(MutableVolumeSet.class), mock(ContainerMetrics.class), mockIcrSender, new ContainerChecksumTreeManager(conf)); }