This is an automated email from the ASF dual-hosted git repository. errose28 pushed a commit to branch HDDS-14496-zdu in repository https://gitbox.apache.org/repos/asf/ozone.git
commit 4ef88e7b2acdd3d3ba0813b7faed6132169b7767 Merge: a9a64dbed31 78784ba9fac Author: Ethan Rose <[email protected]> AuthorDate: Thu Oct 8 18:49:36 2026 -0400 Merge branch 'master' into HDDS-14496-zdu * master: (23 commits) HDDS-15951. Reduce ofs getFileChecksum to a single OM RPC (#11380) HDDS-16746. Local Stream onCommand should drain buffers before commit PutBlock (#11427) HDDS-16781. Fix intermittent ReadChunk count mismatch in TestKeyInputStream (#11443) HDDS-16779. FSO rename of a file to itself should be a no-op (#11440) HDDS-16724. Add support for missing S3 apis to STS (#11413) HDDS-16620. Expose BlockReadCursor as an iterator of streaming ReadBlock ranges (#11442) HDDS-16624. Improve AbortIncompleteMultipartUpload lifecycle validation error message (#11344) HDDS-16367. Update threat model with information about internal protocols (#11185) HDDS-16304. Convert RpcMetrics latency counters to lock-free ConcurrentMutableStat (#11425) HDDS-15681. Switch to io.github.ascopes:protobuf-maven-plugin (#11429) HDDS-16258. Refactor streaming block reads and fix checksum verification for variable-sized chunks (#11302) HDDS-16598. Fix intermittent failure in TestContainerStateMachine (#11319) HDDS-15802. Add integration tests for per-volume push replication (#11146) HDDS-16657. Reject lifecycle Transition actions in PutBucketLifecycleConfiguration (#11430) HDDS-16714. Split kubernetes check by environment (#11406) HDDS-16752. Bump guava to 33.7.2-jre (#11433) HDDS-16581. Bump ozone-runner to 20260922-1 (#11431) HDDS-16611. Use MutableGaugeLong for total count metrics in OMMetrics (#11410) HDDS-16738. Let dependabot ignore jetty minor version upgrades (#11422) HDDS-16683. Upgrade Jetty from 12.0.38 to 12.1.14 (#11424) ... Conflicts: hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneManagerVersion.java hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueHandler.java hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/file/OMFileCreateRequestWithFSO.java .github/dependabot.yml | 2 + .github/workflows/ci.yml | 12 +- CONTRIBUTING.md | 4 +- THREAT_MODEL.md | 46 +- dev-support/ci/kubernetes_suites.sh | 26 + .../apache/hadoop/hdds/scm/OzoneClientConfig.java | 2 +- .../hdds/scm/storage/StreamBlockInputStream.java | 16 +- .../scm/storage/TestStreamBlockInputStream.java | 116 ++- .../org/apache/hadoop/hdds/scm/ScmConfigKeys.java | 30 + .../ContainerCommandResponseBuilders.java | 12 +- .../hdds/scm/storage/ContainerProtocolCalls.java | 3 +- .../org/apache/hadoop/ipc_/metrics/RpcMetrics.java | 41 +- .../org/apache/hadoop/ozone/OzoneConfigKeys.java | 3 + .../apache/hadoop/ozone/OzoneManagerVersion.java | 5 +- .../org/apache/hadoop/ozone/common/Checksum.java | 58 +- .../common/src/main/resources/ozone-default.xml | 65 ++ .../apache/hadoop/ipc_/metrics/TestRpcMetrics.java | 195 +++++ .../apache/hadoop/ozone/common/TestChecksum.java | 63 ++ .../container/common/impl/HddsDispatcher.java | 2 +- .../ozone/container/common/interfaces/Handler.java | 2 +- .../common/transport/server/ratis/LocalStream.java | 6 +- .../ozone/container/keyvalue/BlockReadCursor.java | 146 ++++ .../ozone/container/keyvalue/KeyValueHandler.java | 126 +--- .../container/keyvalue/ReadBlockComputation.java | 159 ---- .../ozone/container/keyvalue/impl/Buffers.java | 5 + .../keyvalue/impl/KeyValueStreamDataChannel.java | 19 + .../server/ratis/ContainerStateMachineTests.java | 78 ++ .../container/keyvalue/TestBlockReadCursor.java | 130 ++++ .../container/keyvalue/TestKeyValueHandler.java | 142 +++- .../impl/TestKeyValueStreamDataChannel.java | 34 + .../hadoop/hdds/server/http/HttpServer2.java | 9 +- .../org/apache/hadoop/hdds/utils/db/Table.java | 22 + .../apache/hadoop/hdds/utils/db/TypedTable.java | 39 +- .../hadoop/hdds/server/http/TestHttpServer2.java | 59 ++ .../hdds/server/http/TestHttpServer2SSL.java | 30 +- .../hadoop/hdds/utils/db/TestTypedTable.java | 56 +- hadoop-hdds/interface-admin/pom.xml | 10 +- hadoop-hdds/interface-client/pom.xml | 30 +- .../src/main/proto/DatanodeClientProtocol.proto | 5 +- hadoop-hdds/interface-server/pom.xml | 30 +- hadoop-hdds/server-scm/pom.xml | 8 +- .../container/replication/ReplicationManager.java | 298 +++++++- .../hadoop/hdds/scm/ha/InterSCMGrpcClient.java | 10 + .../hdds/scm/ha/InterSCMGrpcProtocolService.java | 23 +- .../replication/TestReplicationManager.java | 441 +++++++++++ .../hadoop/hdds/scm/ha/TestInterSCMGrpcClient.java | 28 + .../ozone/shell/keys/ChecksumKeyHandler.java | 5 +- .../hadoop/ozone/client/OzoneClientUtils.java | 23 +- .../client/checksum/BaseFileChecksumHelper.java | 19 +- .../client/checksum/ChecksumHelperFactory.java | 10 +- .../client/checksum/ECFileChecksumHelper.java | 6 +- .../checksum/ReplicatedFileChecksumHelper.java | 6 +- .../ozone/client/protocol/ClientProtocol.java | 8 + .../apache/hadoop/ozone/client/rpc/RpcClient.java | 5 + .../hadoop/ozone/client/TestOzoneClientUtils.java | 12 +- .../client/checksum/FileChecksumBenchmark.java | 9 +- .../client/checksum/TestFileChecksumHelper.java | 26 +- .../security/acl/iam/IamSessionPolicyResolver.java | 10 + .../acl/iam/TestIamSessionPolicyResolver.java | 140 +++- hadoop-ozone/dev-support/checks/kubernetes.sh | 7 + hadoop-ozone/dist/pom.xml | 4 +- .../Ozone - Memory Consumption Metrics.json | 4 +- hadoop-ozone/dist/src/main/k8s/examples/testlib.sh | 14 - hadoop-ozone/dist/src/main/license/bin/LICENSE.txt | 3 +- hadoop-ozone/dist/src/main/license/jar-report.txt | 3 +- .../main/smoketest/security/ozone-secure-sts.robot | 200 ++++- .../org/apache/hadoop/ozone/s3/ProxyServer.java | 28 + .../fs/ozone/AbstractOzoneFileSystemTest.java | 5 + .../ozone/AbstractRootedOzoneFileSystemTest.java | 66 ++ .../hadoop/fs/ozone/OzoneFileSystemTestBase.java | 24 + .../ozone/TestOfsGetFileChecksumRpcBenchmark.java | 314 ++++++++ .../client/rpc/TestContainerStateMachine.java | 47 +- .../rpc/TestContainerStateMachineStream.java | 23 + .../ozone/client/rpc/read/TestKeyInputStream.java | 13 +- .../ozone/client/rpc/read/TestStreamRead.java | 63 ++ .../replication/TestPerVolumePushReplication.java | 553 ++++++++++++++ .../hadoop/ozone/om/TestOMUpgradeFinalization.java | 2 +- .../org/apache/hadoop/ozone/om/TestOmMetrics.java | 53 +- ...stOzoneManagerHAFollowerReadWithAllRunning.java | 19 +- .../om/service/TestOmFsoWriteLockConvoyBench.java | 837 +++++++++++++++++++++ .../om/snapshot/TestOzoneManagerSnapshotAcl.java | 30 +- hadoop-ozone/interface-client/pom.xml | 21 +- hadoop-ozone/interface-storage/pom.xml | 9 +- .../java/org/apache/hadoop/ozone/om/OMMetrics.java | 39 +- .../apache/hadoop/ozone/om/OmMetadataReader.java | 46 +- .../file/OMDirectoryCreateRequestWithFSO.java | 251 ++++-- .../request/file/OMFileCreateRequestWithFSO.java | 335 ++++++--- .../ozone/om/request/file/OMFileRequest.java | 79 +- .../om/request/key/OMKeyCommitRequestWithFSO.java | 608 +++++++++------ .../om/request/key/OMKeyDeleteRequestWithFSO.java | 329 +++++--- .../om/request/key/OMKeyRenameRequestWithFSO.java | 364 +++++---- .../OMLifecycleConfigurationSetRequest.java | 5 +- .../om/request/s3/security/OMSetSecretRequest.java | 2 +- .../om/request/s3/security/S3GetSecretRequest.java | 2 +- .../s3/security/S3RevokeSTSTokenRequest.java | 2 +- .../request/s3/security/S3RevokeSecretRequest.java | 2 +- .../request/s3/security/S3SecretRequestHelper.java | 8 +- .../om/snapshot/diff/SnapshotDiffValueParser.java | 29 + .../hadoop/ozone/om/TestOMMetadataReader.java | 133 ++++ .../hadoop/ozone/om/TestOMMetricsSetters.java | 58 ++ .../file/TestOMFileCreateRequestWithFSO.java | 29 + .../request/key/TestOMKeyDeleteRequestWithFSO.java | 98 +++ .../request/key/TestOMKeyRenameRequestWithFSO.java | 141 ++++ .../s3/security/TestS3GetSecretRequest.java | 30 + .../s3/security/TestS3RevokeSTSTokenRequest.java | 15 + .../s3/security/TestS3SecretRequestHelper.java | 31 +- .../ozone/om/upgrade/TestOMVersionManager.java | 2 +- .../fs/ozone/BasicOzoneClientAdapterImpl.java | 11 +- .../hadoop/fs/ozone/BasicOzoneFileSystem.java | 5 + .../ozone/BasicRootedOzoneClientAdapterImpl.java | 51 +- .../fs/ozone/BasicRootedOzoneFileSystem.java | 5 + .../apache/hadoop/fs/ozone/OzoneClientAdapter.java | 2 + ...sicRootedOzoneClientAdapterGetFileChecksum.java | 199 +++++ .../hadoop/ozone/s3/endpoint/EndpointBase.java | 1 + .../ozone/s3/endpoint/ObjectAttributesHandler.java | 2 + .../s3/endpoint/S3LifecycleConfiguration.java | 63 ++ .../hadoop/ozone/s3/util/S3GActionIamMapper.java | 13 + .../TestS3ActionOverrideForOwnerVerification.java | 143 +++- .../endpoint/TestS3LifecycleConfigurationPut.java | 27 + .../ozone/s3/util/TestS3GActionIamMapper.java | 7 + pom.xml | 14 +- 121 files changed, 7070 insertions(+), 1378 deletions(-) diff --cc hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneManagerVersion.java index 95cec6847cf,f33274dde83..bce3ab2e1e9 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneManagerVersion.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneManagerVersion.java @@@ -68,17 -66,20 +68,20 @@@ public enum OzoneManagerVersion impleme "OzoneManager version that rejects getFileStatus on OBJECT_STORE " + "buckets server-side, so file system clients no longer need the " + "client-side InfoBucket layout check"), -- + LOOKUP_FILE_REJECTS_OBS(15, + "OzoneManager version that rejects lookupFile on OBJECT_STORE buckets " + + "server-side, so file system clients no longer need the " + + "client-side InfoBucket layout check for getFileChecksum"), + ZDU(100, "OzoneManager version that supports zero downtime upgrade"), - FUTURE_VERSION(-1, "Used internally in the client when the server side is " - + " newer and an unknown server version has arrived to the client."); - - public static final OzoneManagerVersion CURRENT = latest(); + UNKNOWN_VERSION(-1, "Used when a version cannot be deserialized to any version recognized by this" + + " component, which may indicate it came from a component in a newer version"); - private static final Map<Integer, OzoneManagerVersion> BY_PROTO_VALUE = + private static final SortedMap<Integer, OzoneManagerVersion> BY_VALUE = Arrays.stream(values()) - .collect(toMap(OzoneManagerVersion::toProtoValue, identity())); + .collect(toMap(OzoneManagerVersion::serialize, identity(), (v1, v2) -> v1, TreeMap::new)); + + public static final OzoneManagerVersion SOFTWARE_VERSION = BY_VALUE.get(BY_VALUE.lastKey()); private final int version; private final String description; diff --cc hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueHandler.java index 24c30c463dd,6a5bf67a66d..721681619c9 --- 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 @@@ -137,7 -132,8 +136,9 @@@ import org.junit.jupiter.api.Assertions import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; + import org.junit.jupiter.params.ParameterizedTest; + import org.junit.jupiter.params.provider.ValueSource; +import org.mockito.ArgumentCaptor; import org.mockito.Mockito; import org.slf4j.Logger; import org.slf4j.LoggerFactory; diff --cc hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java index 09e28c512ec,29f00a10d3c..97324720392 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java @@@ -540,47 -586,28 +590,56 @@@ public class ReplicationManager impleme public void sendThrottledReconstructionCommand(ContainerInfo containerInfo, ReconstructECContainersCommand command) throws CommandTargetOverloadedException, NotLeaderException { - List<DatanodeDetails> targets = command.getTargetDatanodes(); - List<Pair<Integer, DatanodeDetails>> targetWithCmds = - getAvailableDatanodesForReplication(targets); - if (targetWithCmds.isEmpty()) { - metrics.incrECReconstructionCmdsDeferredTotal(); - throw new CommandTargetOverloadedException("No target with capacity " + - "available for reconstruction of " + containerInfo.getContainerID()); - } - DatanodeDetails target = selectAndOptionallyExcludeDatanode( - rmConf.getReconstructionCommandWeight(), targetWithCmds); - command.setApparentVersion(computeVersionForReconstruction(command)); - sendDatanodeCommand(command, containerInfo, target); + globalReconstructionTracker.reserveAndSend( + command.getId(), + command.getMissingContainerIndexes().size(), + containerInfo.getContainerID(), + () -> { + // Pick the least-loaded target datanode that can accept the command. + List<DatanodeDetails> targets = command.getTargetDatanodes(); + List<Pair<Integer, DatanodeDetails>> targetWithCmds = + getAvailableDatanodesForReplication(targets); + if (targetWithCmds.isEmpty()) { + metrics.incrECReconstructionCmdsDeferredTotal(); + throw new CommandTargetOverloadedException("No target with capacity " + + "available for reconstruction of " + containerInfo.getContainerID()); + } + DatanodeDetails target = selectAndOptionallyExcludeDatanode( + rmConf.getReconstructionCommandWeight(), targetWithCmds); ++ command.setApparentVersion(computeVersionForReconstruction(command)); + // sendDatanodeCommand schedules one pending ADD per missing index and + // increments ecReconstructionCmdsSentTotal. + sendDatanodeCommand(command, containerInfo, target); + }); } + private ComponentVersion computeVersionForReconstruction(ReconstructECContainersCommand command) { + List<DatanodeInfo> involved = new ArrayList<>(); + for (ReconstructECContainersCommand.DatanodeDetailsAndReplicaIndex source : command.getSources()) { + involved.add(getDatanodeInfo(source.getDnDetails())); + } + for (DatanodeDetails target : command.getTargetDatanodes()) { + involved.add(getDatanodeInfo(target)); + } + // Reconstruction is handled as a specific type of replication command which uses the same generic logic within the + // version manager as computing a version for full container replication. + return ScmVersionManager.computeVersionForReplication(involved); + } + + private ComponentVersion computeVersionForReplication(DatanodeDetails source, DatanodeDetails target) { + return ScmVersionManager.computeVersionForReplication( + Arrays.asList(getDatanodeInfo(source), getDatanodeInfo(target))); + } + + private DatanodeInfo getDatanodeInfo(DatanodeDetails dnDetails) { + DatanodeInfo datanodeInfo = nodeManager.getNode(dnDetails.getID()); + if (datanodeInfo == null) { + throw new IllegalArgumentException("Datanode " + dnDetails + " not " + + "found in NodeManager. Should not happen"); + } + return datanodeInfo; + } + private DatanodeDetails selectAndOptionallyExcludeDatanode( int additionalCmdCount, List<Pair<Integer, DatanodeDetails>> datanodes) { if (datanodes.isEmpty()) { diff --cc hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java index 5c558bae902,032789d180a..ac6ad3139b8 --- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java +++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java @@@ -61,12 -63,15 +63,17 @@@ import java.util.HashSet import java.util.List; import java.util.Map; import java.util.Set; + import java.util.concurrent.CountDownLatch; + import java.util.concurrent.ExecutorService; + import java.util.concurrent.Executors; + import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; + import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Function; import org.apache.commons.lang3.tuple.Pair; +import org.apache.hadoop.hdds.ComponentVersion; +import org.apache.hadoop.hdds.HDDSVersion; import org.apache.hadoop.hdds.client.ECReplicationConfig; import org.apache.hadoop.hdds.client.RatisReplicationConfig; import org.apache.hadoop.hdds.client.ReplicationConfig; @@@ -1828,94 -1746,437 +1835,528 @@@ public class TestReplicationManager }); } + /** + * Captures the ReplicateContainerCommand SCM sends to a datanode. + */ + private ReplicateContainerCommand captureSentReplicateCommand() { + ArgumentCaptor<SCMCommand<?>> command = + ArgumentCaptor.forClass(SCMCommand.class); + verify(nodeManager).addDatanodeCommand(any(DatanodeID.class), + command.capture()); + return (ReplicateContainerCommand) command.getValue(); + } + + private void mockDatanodeWithApparentVersion( + DatanodeDetails dn, ComponentVersion version) { + DatanodeInfo info = mock(DatanodeInfo.class); + when(info.getLastKnownApparentVersion()).thenReturn(version); + when(nodeManager.getNode(dn.getID())).thenReturn(info); + } + + /** + * Regardless of the command sending path, the replicate command must carry + * the lowest apparent version among the source and target datanodes. This + * case places the lowest version on the source. + */ + @ParameterizedTest + @ValueSource(booleans = {true, false}) + public void testReplicationApparentVersionWhenSourceIsLowest(boolean throttled) + throws CommandTargetOverloadedException, NotLeaderException, + NodeNotFoundException { + assertReplicationCarriesLowestApparentVersion(throttled, true); + } + + /** + * Regardless of the command sending path, the replicate command must carry + * the lowest apparent version among the source and target datanodes. This + * case places the lowest version on the target. + */ + @ParameterizedTest + @ValueSource(booleans = {true, false}) + public void testReplicationApparentVersionWhenTargetIsLowest(boolean throttled) + throws CommandTargetOverloadedException, NotLeaderException, + NodeNotFoundException { + assertReplicationCarriesLowestApparentVersion(throttled, false); + } + + private void assertReplicationCarriesLowestApparentVersion( + boolean throttled, boolean sourceIsLowest) + throws CommandTargetOverloadedException, NotLeaderException, + NodeNotFoundException { + ContainerInfo containerInfo = + ReplicationTestUtil.createContainerInfo(repConfig, 1, + HddsProtos.LifeCycleState.CLOSED, 10, 20); + DatanodeDetails target = MockDatanodeDetails.randomDatanodeDetails(); + DatanodeDetails source = MockDatanodeDetails.randomDatanodeDetails(); + + ComponentVersion lower = HDDSLayoutFeature.STORAGE_SPACE_DISTRIBUTION; + ComponentVersion higher = HDDSVersion.ZDU; + mockDatanodeWithApparentVersion(source, sourceIsLowest ? lower : higher); + mockDatanodeWithApparentVersion(target, sourceIsLowest ? higher : lower); + + if (throttled) { + mockReplicationCommandCounts(dn -> 0, dn -> 0); + replicationManager.sendThrottledReplicationCommand(containerInfo, + Collections.singletonList(source), target, 1); + } else { + replicationManager.sendLowPriorityReplicateContainerCommand(containerInfo, + 1, source, target, clock.millis() + rmConf.getEventTimeout()); + } + + assertEquals(lower, captureSentReplicateCommand().getApparentVersion()); + } + + @Test + public void testReplicationApparentVersionLookupThrowsWhenNodeNotFound() + throws NodeNotFoundException { + ContainerInfo containerInfo = + ReplicationTestUtil.createContainerInfo(repConfig, 1, + HddsProtos.LifeCycleState.CLOSED, 10, 20); + DatanodeDetails target = MockDatanodeDetails.randomDatanodeDetails(); + DatanodeDetails source = MockDatanodeDetails.randomDatanodeDetails(); + + mockDatanodeWithApparentVersion(source, HDDSVersion.SOFTWARE_VERSION); + // SCM has no information for the target. + when(nodeManager.getNode(target.getID())).thenReturn(null); + + // We must not proceed with a replication command for a node we don't know. + assertThrows(IllegalArgumentException.class, () -> + replicationManager.sendLowPriorityReplicateContainerCommand( + containerInfo, 0, source, target, + clock.millis() + rmConf.getEventTimeout())); + } ++ + @Test + public void testReconstructionGlobalLimitDisabledByDefault() + throws IOException, NodeNotFoundException { + assertEquals(0, rmConf.getReconstructionGlobalLimit()); + ReplicationManager rm = createReplicationManager(); + assertEquals(0, rm.getReconstructionInFlightLimit()); + assertFalse(rm.isReconstructionLimitReached()); + mockReplicationCommandCounts(dn -> 0, dn -> 0); + + ContainerInfo container = ReplicationTestUtil.createContainerInfo( + repConfig, 1, HddsProtos.LifeCycleState.CLOSED, 10, 20); + ReconstructECContainersCommand cmd = new ReconstructECContainersCommand( + 1L, Collections.emptyList(), + ImmutableList.of(MockDatanodeDetails.randomDatanodeDetails()), + ECUnderReplicationHandler.integers2ByteString(ImmutableList.of(1)), + (ECReplicationConfig) repConfig); + rm.sendThrottledReconstructionCommand(container, cmd); + assertEquals(1, rm.getInflightReconstructionCount()); + assertFalse(rm.isReconstructionLimitReached()); + } + + @Test + public void testInflightReconstructionLimit() throws IOException, NodeNotFoundException { + rmConf.setReconstructionGlobalLimit(2); + ReplicationManager rm = createReplicationManager(); + assertEquals(2, rm.getReconstructionInFlightLimit()); + assertEquals(0, rm.getInflightReconstructionCount()); + assertFalse(rm.isReconstructionLimitReached()); + + mockReplicationCommandCounts(dn -> 0, dn -> 0); + + ContainerInfo container = ReplicationTestUtil.createContainerInfo( + repConfig, 1, HddsProtos.LifeCycleState.CLOSED, 10, 20); + + // Send one reconstruction command with 2 fragments + ReconstructECContainersCommand cmd1 = new ReconstructECContainersCommand( + 1L, Collections.emptyList(), + ImmutableList.of(MockDatanodeDetails.randomDatanodeDetails(), + MockDatanodeDetails.randomDatanodeDetails()), + ECUnderReplicationHandler.integers2ByteString(ImmutableList.of(1, 2)), + (ECReplicationConfig) repConfig); + + rm.sendThrottledReconstructionCommand(container, cmd1); + assertEquals(1, rm.getInflightReconstructionCount()); + assertFalse(rm.isReconstructionLimitReached()); + + // Send another reconstruction command with 1 fragment + ReconstructECContainersCommand cmd2 = new ReconstructECContainersCommand( + 2L, Collections.emptyList(), + ImmutableList.of(MockDatanodeDetails.randomDatanodeDetails()), + ECUnderReplicationHandler.integers2ByteString(ImmutableList.of(3)), + (ECReplicationConfig) repConfig); + rm.sendThrottledReconstructionCommand(container, cmd2); + assertEquals(2, rm.getInflightReconstructionCount()); + assertTrue(rm.isReconstructionLimitReached()); + + // Complete one fragment of cmd1 + ContainerReplicaOp op1 = new ContainerReplicaOp( + ContainerReplicaOp.PendingOpType.ADD, + cmd1.getTargetDatanodes().get(0), 1, cmd1, Long.MAX_VALUE, 0); + rm.opCompleted(op1, container.containerID(), false); + // Still 2 because cmd1 is not fully finished + assertEquals(2, rm.getInflightReconstructionCount()); + + // Complete second fragment of cmd1 + ContainerReplicaOp op2 = new ContainerReplicaOp( + ContainerReplicaOp.PendingOpType.ADD, + cmd1.getTargetDatanodes().get(1), 2, cmd1, Long.MAX_VALUE, 0); + rm.opCompleted(op2, container.containerID(), false); + // Now 1 + assertEquals(1, rm.getInflightReconstructionCount()); + assertFalse(rm.isReconstructionLimitReached()); + + // Complete cmd2 + ContainerReplicaOp op3 = new ContainerReplicaOp( + ContainerReplicaOp.PendingOpType.ADD, + cmd2.getTargetDatanodes().get(0), 3, cmd2, Long.MAX_VALUE, 0); + rm.opCompleted(op3, container.containerID(), false); + assertEquals(0, rm.getInflightReconstructionCount()); + } + + @Test + public void testSendReconstructionCommandRejectedWhenGlobalLimitReached() + throws IOException, NodeNotFoundException { + rmConf.setReconstructionGlobalLimit(2); + ReplicationManager rm = createReplicationManager(); + mockReplicationCommandCounts(dn -> 0, dn -> 0); + + ContainerInfo container = ReplicationTestUtil.createContainerInfo( + repConfig, 1, HddsProtos.LifeCycleState.CLOSED, 10, 20); + + ReconstructECContainersCommand cmd1 = new ReconstructECContainersCommand( + 1L, Collections.emptyList(), + ImmutableList.of(MockDatanodeDetails.randomDatanodeDetails()), + ECUnderReplicationHandler.integers2ByteString(ImmutableList.of(1)), + (ECReplicationConfig) repConfig); + ReconstructECContainersCommand cmd2 = new ReconstructECContainersCommand( + 2L, Collections.emptyList(), + ImmutableList.of(MockDatanodeDetails.randomDatanodeDetails()), + ECUnderReplicationHandler.integers2ByteString(ImmutableList.of(2)), + (ECReplicationConfig) repConfig); + ReconstructECContainersCommand cmd3 = new ReconstructECContainersCommand( + 3L, Collections.emptyList(), + ImmutableList.of(MockDatanodeDetails.randomDatanodeDetails()), + ECUnderReplicationHandler.integers2ByteString(ImmutableList.of(3)), + (ECReplicationConfig) repConfig); + + rm.sendThrottledReconstructionCommand(container, cmd1); + rm.sendThrottledReconstructionCommand(container, cmd2); + assertEquals(2, rm.getInflightReconstructionCount()); + + assertThrows(CommandTargetOverloadedException.class, + () -> rm.sendThrottledReconstructionCommand(container, cmd3)); + assertEquals(2, rm.getInflightReconstructionCount()); + } + + @Test + public void testNotifyStatusChangedClearsReconstructionCounters() + throws IOException, NodeNotFoundException { + rmConf.setReconstructionGlobalLimit(10); + ReplicationManager rm = createReplicationManager(); + mockReplicationCommandCounts(dn -> 0, dn -> 0); + enableProcessAll(); + + ContainerInfo container = ReplicationTestUtil.createContainerInfo( + repConfig, 1, HddsProtos.LifeCycleState.CLOSED, 10, 20); + ReconstructECContainersCommand cmd = new ReconstructECContainersCommand( + 1L, Collections.emptyList(), + ImmutableList.of(MockDatanodeDetails.randomDatanodeDetails()), + ECUnderReplicationHandler.integers2ByteString(ImmutableList.of(1)), + (ECReplicationConfig) repConfig); + rm.sendThrottledReconstructionCommand(container, cmd); + assertEquals(1, rm.getInflightReconstructionCount()); + + when(scmContext.isLeaderReady()).thenReturn(false); + rm.notifyStatusChanged(); + when(scmContext.isLeaderReady()).thenReturn(true); + when(scmContext.isInSafeMode()).thenReturn(false); + rm.notifyStatusChanged(); + + assertEquals(0, rm.getInflightReconstructionCount()); + } + + @Test + public void testInflightReconstructionCountNotNegativeAfterFailoverClear() + throws IOException, NodeNotFoundException { + rmConf.setReconstructionGlobalLimit(10); + ReplicationManager rm = createReplicationManager(); + mockReplicationCommandCounts(dn -> 0, dn -> 0); + enableProcessAll(); + + ContainerInfo container = ReplicationTestUtil.createContainerInfo( + repConfig, 1, HddsProtos.LifeCycleState.CLOSED, 10, 20); + ReconstructECContainersCommand cmd = new ReconstructECContainersCommand( + 1L, Collections.emptyList(), + ImmutableList.of(MockDatanodeDetails.randomDatanodeDetails(), + MockDatanodeDetails.randomDatanodeDetails()), + ECUnderReplicationHandler.integers2ByteString(ImmutableList.of(1, 2)), + (ECReplicationConfig) repConfig); + rm.sendThrottledReconstructionCommand(container, cmd); + assertEquals(1, rm.getInflightReconstructionCount()); + + when(scmContext.isLeaderReady()).thenReturn(false); + rm.notifyStatusChanged(); + when(scmContext.isLeaderReady()).thenReturn(true); + when(scmContext.isInSafeMode()).thenReturn(false); + rm.notifyStatusChanged(); + assertEquals(0, rm.getInflightReconstructionCount()); + + ContainerReplicaOp op1 = new ContainerReplicaOp( + ContainerReplicaOp.PendingOpType.ADD, + cmd.getTargetDatanodes().get(0), 1, cmd, Long.MAX_VALUE, 0); + ContainerReplicaOp op2 = new ContainerReplicaOp( + ContainerReplicaOp.PendingOpType.ADD, + cmd.getTargetDatanodes().get(1), 2, cmd, Long.MAX_VALUE, 0); + + rm.opCompleted(op1, container.containerID(), false); + assertEquals(0, rm.getInflightReconstructionCount()); + rm.opCompleted(op2, container.containerID(), false); + assertEquals(0, rm.getInflightReconstructionCount()); + rm.opCompleted(op1, container.containerID(), false); + assertEquals(0, rm.getInflightReconstructionCount()); + } + + @Test + public void testReconstructionConfigValidation() { + ReplicationManager.ReplicationManagerConfiguration config = + new ReplicationManager.ReplicationManagerConfiguration(); + + config.setEcDecommissionReconstructionLoadFactor(-0.1); + assertThrows(IllegalArgumentException.class, config::validate); + + config.setEcDecommissionReconstructionLoadFactor(Double.NaN); + assertThrows(IllegalArgumentException.class, config::validate); + + config.setEcDecommissionReconstructionLoadFactor(1.1); + assertThrows(IllegalArgumentException.class, config::validate); + + config.setEcDecommissionReconstructionLoadFactor(0.9); + config.setReconstructionGlobalLimit(-1); + assertThrows(IllegalArgumentException.class, config::validate); + + config.setReconstructionGlobalLimit(0); + config.setInflightReplicationLimitFactor(Double.NaN); + assertThrows(IllegalArgumentException.class, config::validate); + + config.setInflightReplicationLimitFactor(0.75); + config.validate(); + } + + @Test + public void testReconstructionGlobalLimitEnforcedConcurrently() + throws InterruptedException, NodeNotFoundException, IOException { + rmConf.setReconstructionGlobalLimit(2); + ReplicationManager rm = createReplicationManager(); + mockReplicationCommandCounts(dn -> 0, dn -> 0); + + ContainerInfo container = ReplicationTestUtil.createContainerInfo( + repConfig, 1, HddsProtos.LifeCycleState.CLOSED, 10, 20); + int threadCount = 8; + int attemptsPerThread = 10; + ExecutorService executor = Executors.newFixedThreadPool(threadCount); + CountDownLatch startLatch = new CountDownLatch(1); + CountDownLatch doneLatch = new CountDownLatch(threadCount); + AtomicInteger accepted = new AtomicInteger(); + AtomicInteger rejected = new AtomicInteger(); + + for (int t = 0; t < threadCount; t++) { + final long commandIdBase = t + 100; + executor.submit(() -> { + try { + startLatch.await(); + for (int i = 0; i < attemptsPerThread; i++) { + ReconstructECContainersCommand cmd = new ReconstructECContainersCommand( + commandIdBase + i, Collections.emptyList(), + ImmutableList.of(MockDatanodeDetails.randomDatanodeDetails()), + ECUnderReplicationHandler.integers2ByteString(ImmutableList.of(1)), + (ECReplicationConfig) repConfig); + try { + rm.sendThrottledReconstructionCommand(container, cmd); + accepted.incrementAndGet(); + } catch (CommandTargetOverloadedException | NotLeaderException e) { + rejected.incrementAndGet(); + } + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + doneLatch.countDown(); + } + }); + } + + try { + startLatch.countDown(); + assertTrue(doneLatch.await(30, TimeUnit.SECONDS)); + assertEquals(2, rm.getInflightReconstructionCount()); + assertEquals(2, accepted.get()); + assertTrue(rejected.get() > 0); + } finally { + executor.shutdown(); + } + } + + @Test + public void testReconstructionMapRegisteredBeforePendingOpsScheduled() + throws IOException, NodeNotFoundException { + rmConf.setReconstructionGlobalLimit(10); + ReplicationManager rm = createReplicationManager(); + mockReplicationCommandCounts(dn -> 0, dn -> 0); + + ContainerInfo container = ReplicationTestUtil.createContainerInfo( + repConfig, 1, HddsProtos.LifeCycleState.CLOSED, 10, 20); + ReconstructECContainersCommand cmd = new ReconstructECContainersCommand( + 1L, Collections.emptyList(), + ImmutableList.of(MockDatanodeDetails.randomDatanodeDetails(), + MockDatanodeDetails.randomDatanodeDetails()), + ECUnderReplicationHandler.integers2ByteString(ImmutableList.of(1, 2)), + (ECReplicationConfig) repConfig); + + rm.sendThrottledReconstructionCommand(container, cmd); + assertEquals(2, rm.getReconstructionPendingFragmentCount(cmd.getId())); + assertEquals(1, rm.getInflightReconstructionCount()); + + ContainerReplicaOp op = new ContainerReplicaOp( + ContainerReplicaOp.PendingOpType.ADD, + cmd.getTargetDatanodes().get(0), 1, cmd, Long.MAX_VALUE, 0); + rm.opCompleted(op, container.containerID(), false); + assertEquals(1, rm.getReconstructionPendingFragmentCount(cmd.getId())); + assertEquals(1, rm.getInflightReconstructionCount()); + } + + @Test + public void testInflightReconstructionCountDecrementsWhenAddOpExpires() + throws IOException, NodeNotFoundException { + rmConf.setReconstructionGlobalLimit(10); + ReplicationManager rm = createReplicationManager(); + mockReplicationCommandCounts(dn -> 0, dn -> 0); + + ContainerInfo container = ReplicationTestUtil.createContainerInfo( + repConfig, 1, HddsProtos.LifeCycleState.CLOSED, 10, 20); + ReconstructECContainersCommand cmd = new ReconstructECContainersCommand( + 1L, Collections.emptyList(), + ImmutableList.of(MockDatanodeDetails.randomDatanodeDetails(), + MockDatanodeDetails.randomDatanodeDetails()), + ECUnderReplicationHandler.integers2ByteString(ImmutableList.of(1, 2)), + (ECReplicationConfig) repConfig); + + rm.sendThrottledReconstructionCommand(container, cmd); + assertEquals(2, rm.getReconstructionPendingFragmentCount(cmd.getId())); + assertEquals(1, rm.getInflightReconstructionCount()); + + ContainerReplicaOp firstAdd = new ContainerReplicaOp( + ContainerReplicaOp.PendingOpType.ADD, + cmd.getTargetDatanodes().get(0), 1, cmd, Long.MAX_VALUE, 0); + rm.opCompleted(firstAdd, container.containerID(), true); + assertEquals(1, rm.getReconstructionPendingFragmentCount(cmd.getId())); + assertEquals(1, rm.getInflightReconstructionCount()); + + ContainerReplicaOp secondAdd = new ContainerReplicaOp( + ContainerReplicaOp.PendingOpType.ADD, + cmd.getTargetDatanodes().get(1), 2, cmd, Long.MAX_VALUE, 0); + rm.opCompleted(secondAdd, container.containerID(), true); + assertEquals(0, rm.getReconstructionPendingFragmentCount(cmd.getId())); + assertEquals(0, rm.getInflightReconstructionCount()); + } + + @Test + public void testReconstructionWithNoFragmentsDoesNotReserveSlot() + throws NodeNotFoundException, IOException { + rmConf.setReconstructionGlobalLimit(10); + ReplicationManager rm = createReplicationManager(); + mockReplicationCommandCounts(dn -> 0, dn -> 0); + + ContainerInfo container = ReplicationTestUtil.createContainerInfo( + repConfig, 1, HddsProtos.LifeCycleState.CLOSED, 10, 20); + ReconstructECContainersCommand cmd = new ReconstructECContainersCommand( + 1L, Collections.emptyList(), + Collections.emptyList(), + ECUnderReplicationHandler.integers2ByteString(Collections.emptyList()), + (ECReplicationConfig) repConfig); + + assertThrows(IllegalArgumentException.class, + () -> rm.sendThrottledReconstructionCommand(container, cmd)); + assertEquals(0, rm.getInflightReconstructionCount()); + assertEquals(0, rm.getReconstructionPendingFragmentCount(cmd.getId())); + verify(nodeManager, never()).addDatanodeCommand(any(), any()); + } + + @Test + public void testReconstructionSlotReleasedAndMapClearedOnSendFailure() + throws NodeNotFoundException, IOException { + rmConf.setReconstructionGlobalLimit(10); + ReplicationManager rm = createReplicationManager(); + + int limit = rm.getConfig().getDatanodeReplicationLimit(); + int reconstructionWeight = rm.getConfig().getReconstructionCommandWeight(); + int reconstructionCount = 2; + int replicationCount = limit - reconstructionCount * reconstructionWeight; + mockReplicationCommandCounts(any -> replicationCount, any -> reconstructionCount); + + ContainerInfo container = ReplicationTestUtil.createContainerInfo( + repConfig, 1, HddsProtos.LifeCycleState.CLOSED, 10, 20); + ReconstructECContainersCommand cmd = createReconstructionCommand( + container, MockDatanodeDetails.randomDatanodeDetails(), + MockDatanodeDetails.randomDatanodeDetails()); + + assertThrows(CommandTargetOverloadedException.class, + () -> rm.sendThrottledReconstructionCommand(container, cmd)); + assertEquals(0, rm.getInflightReconstructionCount()); + assertEquals(0, rm.getReconstructionPendingFragmentCount(cmd.getId())); + } + + @Test + public void testNotifyStatusChangedBlocksUntilReconstructionSendCompletes() + throws Exception { + rmConf.setReconstructionGlobalLimit(10); + ReplicationManager rm = createReplicationManager(); + mockReplicationCommandCounts(dn -> 0, dn -> 0); + enableProcessAll(); + + CountDownLatch sendReachedAddCommand = new CountDownLatch(1); + CountDownLatch releaseSend = new CountDownLatch(1); + doAnswer(invocation -> { + sendReachedAddCommand.countDown(); + try { + releaseSend.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException(e); + } + return null; + }).when(nodeManager).addDatanodeCommand(any(), any()); + + ContainerInfo container = ReplicationTestUtil.createContainerInfo( + repConfig, 1, HddsProtos.LifeCycleState.CLOSED, 10, 20); + ReconstructECContainersCommand cmd = new ReconstructECContainersCommand( + 1L, Collections.emptyList(), + ImmutableList.of(MockDatanodeDetails.randomDatanodeDetails()), + ECUnderReplicationHandler.integers2ByteString(ImmutableList.of(1)), + (ECReplicationConfig) repConfig); + + Thread sender = new Thread(() -> { + try { + rm.sendThrottledReconstructionCommand(container, cmd); + } catch (CommandTargetOverloadedException | NotLeaderException e) { + throw new RuntimeException(e); + } + }); + sender.start(); + assertTrue(sendReachedAddCommand.await(30, TimeUnit.SECONDS)); + assertEquals(1, rm.getInflightReconstructionCount()); + assertEquals(1, rm.getReconstructionPendingFragmentCount(cmd.getId())); + + AtomicBoolean notifyCompleted = new AtomicBoolean(false); + Thread notifyThread = new Thread(() -> { + rm.notifyStatusChanged(); + notifyCompleted.set(true); + }); + notifyThread.start(); + notifyThread.join(500); + assertFalse(notifyCompleted.get(), + "notifyStatusChanged must wait for in-flight reconstruction send"); + assertEquals(1, rm.getInflightReconstructionCount()); + + releaseSend.countDown(); + sender.join(TimeUnit.SECONDS.toMillis(30)); + notifyThread.join(TimeUnit.SECONDS.toMillis(30)); + assertTrue(notifyCompleted.get()); + assertTrue(containerReplicaPendingOps.getPendingOps(container.containerID()) + .isEmpty()); + assertEquals(0, rm.getInflightReconstructionCount()); + } } diff --cc hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/replication/TestPerVolumePushReplication.java index 00000000000,581983b2507..a0b27723d5c mode 000000,100644..100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/replication/TestPerVolumePushReplication.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/replication/TestPerVolumePushReplication.java @@@ -1,0 -1,552 +1,553 @@@ + /* + * 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.replication; + + import static java.nio.charset.StandardCharsets.UTF_8; + import static java.util.Collections.emptySet; + import static java.util.Collections.singleton; + import static java.util.Collections.singletonList; + import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_CONTAINER_REPORT_INTERVAL; + import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_HEARTBEAT_INTERVAL; + import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_NODE_REPORT_INTERVAL; + import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_PIPELINE_REPORT_INTERVAL; + import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerDataProto.State.CLOSED; + import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeOperationalState.DECOMMISSIONED; + import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeOperationalState.IN_SERVICE; + import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeState.DEAD; + import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor.THREE; + import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_DATANODE_ADMIN_MONITOR_INTERVAL; + import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_DEADNODE_INTERVAL; + import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_HEARTBEAT_PROCESS_INTERVAL; + import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_STALENODE_INTERVAL; + import static org.apache.hadoop.hdds.scm.node.NodeTestUtil.getDNHostAndPort; + import static org.apache.hadoop.hdds.scm.node.NodeTestUtil.waitForDnToReachHealthState; + import static org.apache.hadoop.hdds.scm.node.NodeTestUtil.waitForDnToReachOpState; + import static org.apache.hadoop.hdds.scm.pipeline.MockPipeline.createPipeline; + import static org.apache.hadoop.hdds.scm.storage.ContainerProtocolCalls.createContainer; + import static org.apache.hadoop.ozone.container.OzoneTestHelper.waitForContainerClose; ++import static org.apache.hadoop.ozone.container.common.ContainerTestUtils.getReplicateContainerCommand; + import static org.assertj.core.api.Assertions.assertThat; + + import java.io.IOException; + import java.time.Duration; + import java.util.List; + import java.util.Set; + import java.util.concurrent.ThreadPoolExecutor; + import java.util.concurrent.TimeUnit; + import java.util.concurrent.TimeoutException; + import java.util.concurrent.atomic.AtomicLong; + import java.util.stream.Collectors; + import org.apache.hadoop.hdds.HddsConfigKeys; + import org.apache.hadoop.hdds.client.ECReplicationConfig; + import org.apache.hadoop.hdds.client.RatisReplicationConfig; + import org.apache.hadoop.hdds.client.ReplicationConfig; + import org.apache.hadoop.hdds.conf.OzoneConfiguration; + import org.apache.hadoop.hdds.conf.StorageUnit; + import org.apache.hadoop.hdds.protocol.DatanodeDetails; + import org.apache.hadoop.hdds.protocol.DatanodeID; + import org.apache.hadoop.hdds.scm.ScmConfigKeys; + import org.apache.hadoop.hdds.scm.XceiverClientFactory; + import org.apache.hadoop.hdds.scm.XceiverClientManager; + import org.apache.hadoop.hdds.scm.XceiverClientSpi; + import org.apache.hadoop.hdds.scm.cli.ContainerOperationClient; + import org.apache.hadoop.hdds.scm.container.ContainerID; + import org.apache.hadoop.hdds.scm.container.ContainerInfo; + import org.apache.hadoop.hdds.scm.container.ContainerManager; + import org.apache.hadoop.hdds.scm.container.ContainerReplica; + import org.apache.hadoop.hdds.scm.container.replication.ReplicationManager.ReplicationManagerConfiguration; + import org.apache.hadoop.hdds.scm.node.NodeManager; + import org.apache.hadoop.hdds.scm.pipeline.Pipeline; + import org.apache.hadoop.hdds.scm.pipeline.PipelineManager; + import org.apache.hadoop.hdds.scm.server.StorageContainerManager; + import org.apache.hadoop.hdds.utils.IOUtils; + import org.apache.hadoop.ozone.DataTestUtil; + import org.apache.hadoop.ozone.HddsDatanodeService; + import org.apache.hadoop.ozone.MiniOzoneCluster; + import org.apache.hadoop.ozone.OzoneConfigKeys; + import org.apache.hadoop.ozone.UniformDatanodesFactory; + import org.apache.hadoop.ozone.client.OzoneBucket; + import org.apache.hadoop.ozone.client.OzoneClient; + import org.apache.hadoop.ozone.client.OzoneKeyDetails; + import org.apache.hadoop.ozone.container.common.interfaces.Container; + import org.apache.hadoop.ozone.container.common.statemachine.DatanodeConfiguration; + import org.apache.hadoop.ozone.container.common.statemachine.DatanodeStateMachine; + import org.apache.hadoop.ozone.container.common.statemachine.StateContext; + 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.dn.DatanodeTestUtils; + import org.apache.hadoop.ozone.protocol.commands.ReplicateContainerCommand; + import org.apache.ozone.test.GenericTestUtils; + import org.junit.jupiter.api.AfterAll; + import org.junit.jupiter.api.BeforeAll; + import org.junit.jupiter.api.MethodOrderer; + import org.junit.jupiter.api.Order; + import org.junit.jupiter.api.Test; + import org.junit.jupiter.api.TestInstance; + import org.junit.jupiter.api.TestMethodOrder; + + /** + * Integration tests for per-volume push replication thread pools (HDDS-15412). + * + * <p>The tests share one cluster and run in {@link Order} sequence, so each one restores whatever it broke + * (stopped datanode, failed volume) before returning. {@link #selectHealthyDatanode} additionally filters out + * stopped datanodes so a leaked shutdown cannot silently hand a later test a dead node. + */ + @TestInstance(TestInstance.Lifecycle.PER_CLASS) + @TestMethodOrder(MethodOrderer.OrderAnnotation.class) + class TestPerVolumePushReplication { + + private static final AtomicLong CONTAINER_ID = new AtomicLong(1_000_000L); + private static final int DATA_VOLUMES = 2; + private static final int DATANODE_COUNT = 7; + private static final int PER_VOLUME_STREAMS = 1; + private static final String VOLUME = "vol1"; + private static final String BUCKET = "bucket1"; + private static final RatisReplicationConfig RATIS_THREE = RatisReplicationConfig.getInstance(THREE); + private static final ECReplicationConfig EC_REP = new ECReplicationConfig(3, 2); + + private MiniOzoneCluster cluster; + private XceiverClientFactory clientFactory; + private OzoneClient client; + private OzoneBucket bucket; + + @BeforeAll + void setUp() throws Exception { + OzoneConfiguration conf = createConfig(); + cluster = newCluster(conf, DATANODE_COUNT); + cluster.waitForClusterToBeReady(); + clientFactory = new XceiverClientManager(conf); + // Keep the client open for the lifetime of the class: the bucket handle below delegates to it, and all + // three tests write keys through that handle. + client = cluster.newClient(); + bucket = DataTestUtil.createVolumeAndBucket(client, VOLUME, BUCKET); + } + + @AfterAll + void tearDown() { + IOUtils.closeQuietly(client, clientFactory, cluster); + } + + @Order(1) + @Test + void testPushAndScmReplicationWithPerVolumeEnabled() throws Exception { + HddsDatanodeService sourceDn = selectHealthyDatanode(0); + DatanodeDetails source = sourceDn.getDatanodeDetails(); + DatanodeDetails target = selectOtherHealthyNode(source); + long containerId = createClosedContainer(clientFactory, source); + + assertVolumePools(sourceDn, DATA_VOLUMES, PER_VOLUME_STREAMS); + + // Assert the push was dispatched to the pool of the volume holding the container, not the global pool. + // Without this the test would also pass with hdds.datanode.replication.per.volume.enabled=false. + HddsVolume containerVolume = getContainer(cluster, source, containerId).getContainerData().getVolume(); + ThreadPoolExecutor volumePool = volumePoolOf(sourceDn, containerVolume); + long completedBefore = volumePool.getCompletedTaskCount(); + - ReplicateContainerCommand cmd = ReplicateContainerCommand.toTarget(containerId, target); ++ ReplicateContainerCommand cmd = getReplicateContainerCommand(containerId, target); + queuePushAndWaitForContainer(cluster, cmd, source, target, containerId); + GenericTestUtils.waitFor(() -> volumePool.getCompletedTaskCount() > completedBefore, 100, 30000); + + DataTestUtil.createKey(bucket, "pushKey1", RATIS_THREE, "data".getBytes(UTF_8)); + OzoneKeyDetails keyDetails = bucket.getKey("pushKey1"); + long scmContainerId = keyDetails.getOzoneKeyLocations().get(0).getContainerID(); + waitForContainerClose(cluster, scmContainerId); + + // SCM learns about replicas asynchronously via ICR, so settle on the full replica set before picking one + // to stop; otherwise the iterator below can be empty. + StorageContainerManager scm = cluster.getStorageContainerManager(); + ContainerManager containerManager = scm.getContainerManager(); + ContainerID scmContainer = ContainerID.valueOf(scmContainerId); + waitForReplicas(containerManager, scmContainer, 3); + Set<ContainerReplica> replicas = containerManager.getContainerReplicas(scmContainer); + DatanodeDetails replicaDn = replicas.iterator().next().getDatanodeDetails(); + + cluster.shutdownHddsDatanode(replicaDn); + try { + // SCM drops a dead node's replicas only once it is declared DEAD, so waiting for a count of 3 right after + // the shutdown would be satisfied by the stale replica on its very first poll. Require 3 replicas none of + // which sit on the stopped node, which only re-replication can produce. + waitForDnToReachHealthState(scm.getScmNodeManager(), replicaDn, DEAD); + GenericTestUtils.waitFor(() -> { + Set<ContainerReplica> current = replicasOf(containerManager, scmContainer); + return current.size() == 3 + && current.stream().noneMatch(r -> r.getDatanodeDetails().equals(replicaDn)); + }, 500, 60000); + } finally { + // Restore the cluster so the later tests see all DATANODE_COUNT nodes. + cluster.restartHddsDatanode(replicaDn, true); + } + } + + @Order(2) + @Test + void testHealthyVolumeReplicationAfterVolumeFailure() throws Exception { + HddsDatanodeService sourceDn = selectHealthyDatanode(1); + DatanodeDetails source = sourceDn.getDatanodeDetails(); + DatanodeDetails target = selectOtherHealthyNode(source); + MutableVolumeSet volSet = sourceDn.getDatanodeStateMachine().getContainer().getVolumeSet(); + HddsVolume vol0 = (HddsVolume) volSet.getVolumesList().get(0); + HddsVolume vol1 = (HddsVolume) volSet.getVolumesList().get(1); + + long containerOnVol0 = findOrCreateContainerOnVolume(cluster, clientFactory, source, vol0); + long containerOnVol1 = findOrCreateContainerOnVolume(cluster, clientFactory, source, vol1); + + assertVolumePools(sourceDn, DATA_VOLUMES, PER_VOLUME_STREAMS); + + try { + triggerAndWaitForVolumeFailure(volSet, vol0); + waitForVolumePoolState(sourceDn, vol0, vol1); + + // The healthy volume keeps its own pool and keeps serving pushes from it. + ThreadPoolExecutor healthyPool = volumePoolOf(sourceDn, vol1); + long completedBefore = healthyPool.getCompletedTaskCount(); - ReplicateContainerCommand cmd = ReplicateContainerCommand.toTarget(containerOnVol1, target); ++ ReplicateContainerCommand cmd = getReplicateContainerCommand(containerOnVol1, target); + queuePushAndWaitForContainer(cluster, cmd, source, target, containerOnVol1); + GenericTestUtils.waitFor(() -> healthyPool.getCompletedTaskCount() > completedBefore, 100, 30000); + assertThat(volSet.getFailedVolumesList()).hasSize(1); + + // Task routes via global pool fallback (HDDS-15327); replication fails on bad volume. - ReplicateContainerCommand failedVolCmd = ReplicateContainerCommand.toTarget(containerOnVol0, target); ++ ReplicateContainerCommand failedVolCmd = getReplicateContainerCommand(containerOnVol0, target); + ReplicationSupervisor supervisor = sourceDn.getDatanodeStateMachine().getSupervisor(); + // Scope the counter to push replication: the cluster-wide counter also moves for unrelated + // SCM-driven work on this datanode and would satisfy the wait on its own. + long previousFailures = supervisor.getReplicationFailureCount(ReplicationTask.METRIC_NAME); + queuePushAndWaitForFailure(cluster, failedVolCmd, source, supervisor, previousFailures); + assertThat(supervisor.getReplicationFailureCount(ReplicationTask.METRIC_NAME)) + .isGreaterThanOrEqualTo(previousFailures + 1); + assertThat(hasContainer(cluster, target, containerOnVol0)).isFalse(); + } finally { + // Must run even if an assertion above fails: the volume dir is left read-only otherwise, which breaks + // the next test and stops the cluster from cleaning up its base dir. + DatanodeTestUtils.restoreBadVolume(vol0); + // restoreBadVolume only flips the directory permission. vol0 stays in the failed volume map and its + // pool is not recreated, so this datanode would keep serving one volume for the rest of the class and + // the decommission test would fail its two-pool assertion whenever it happens to pick this node. + // Restarting rescans the data dirs and brings both volumes and both pools back. + cluster.restartHddsDatanode(source, true); + } + } + + @Order(3) + @Test + void testDecommissionWithPerVolumePools() throws Exception { + try (ContainerOperationClient scmClient = new ContainerOperationClient(cluster.getConf())) { + StorageContainerManager scm = cluster.getStorageContainerManager(); + NodeManager nm = scm.getScmNodeManager(); + ContainerManager cm = scm.getContainerManager(); + PipelineManager pm = scm.getPipelineManager(); + + generateData(bucket, 20, "decomKey", RATIS_THREE); + generateData(bucket, 20, "decomEcKey", EC_REP); + + ContainerInfo ratisContainer = waitForKeyContainer(bucket, cm, "decomKey0", 3); + ContainerInfo ecContainer = waitForKeyContainer(bucket, cm, "decomEcKey0", 5); + Pipeline ratisPipeline = pm.getPipeline(ratisContainer.getPipelineID()); + Pipeline ecPipeline = pm.getPipeline(ecContainer.getPipelineID()); + + DatanodeID dnId = ratisPipeline.getNodes().stream() + .filter(node -> ecPipeline.getNodes().contains(node)) + .findFirst() + .orElseThrow(() -> new AssertionError("no intersecting datanode found")) + .getID(); + DatanodeDetails toDecommission = nm.getNode(dnId); + HddsDatanodeService decommissioningDn = cluster.getHddsDatanode(toDecommission); + + scmClient.decommissionNodes(singletonList(getDNHostAndPort(toDecommission)), false); + waitForDnToReachOpState(nm, toDecommission, DECOMMISSIONED); + + // Per-volume behaviour that the existing decommission coverage cannot catch: SCM's operational state + // change reaches the datanode as a SetNodeOperationalStateCommand, which drives + // ReplicationSupervisor.nodeStateUpdated -> resize, scaling every per-volume pool by the + // out-of-service factor. Only an integration test exercises that whole path. + ReplicationSupervisor supervisor = decommissioningDn.getDatanodeStateMachine().getSupervisor(); + int scaledStreams = supervisor.getReplicationConfig().scaleOutOfServiceLimit(PER_VOLUME_STREAMS); + assertThat(scaledStreams).isGreaterThan(PER_VOLUME_STREAMS); + waitForVolumePoolSize(decommissioningDn, scaledStreams); + + waitForContainerReplicas(cm, ratisContainer, 4); + waitForContainerReplicas(cm, ecContainer, 6); + + cluster.shutdownHddsDatanode(toDecommission); + waitForDnToReachHealthState(nm, toDecommission, DEAD); + + waitForContainerReplicas(cm, ratisContainer, 3); + waitForContainerReplicas(cm, ecContainer, 5); + + DataTestUtil.createKey(bucket, "sanityKey", RATIS_THREE, "still healthy".getBytes(UTF_8)); + } + } + + private HddsDatanodeService selectHealthyDatanode(int indexAmongHealthy) { + List<HddsDatanodeService> healthy = cluster.getHddsDatanodes().stream() + .filter(this::isHealthyDatanode) + .collect(Collectors.toList()); + if (indexAmongHealthy >= healthy.size()) { + throw new AssertionError("not enough healthy datanodes: requested index " + + indexAmongHealthy + ", found " + healthy.size()); + } + return healthy.get(indexAmongHealthy); + } + + private DatanodeDetails selectOtherHealthyNode(DatanodeDetails source) { + return cluster.getHddsDatanodes().stream() + .filter(this::isHealthyDatanode) + .map(HddsDatanodeService::getDatanodeDetails) + .filter(dn -> !dn.equals(source)) + .findAny() + .orElseThrow(() -> new AssertionError("no target datanode found")); + } + + private boolean isHealthyDatanode(HddsDatanodeService datanode) { + // A stopped datanode keeps its DatanodeStateMachine and stays IN_SERVICE in getHddsDatanodes(), so the + // op-state and volume checks below cannot tell it from a live one. Without this guard a leaked shutdown + // in an earlier test would hand this one a dead node and the push waits would simply time out. + if (datanode.isStopped()) { + return false; + } + if (datanode.getDatanodeDetails().getPersistedOpState() != IN_SERVICE) { + return false; + } + MutableVolumeSet volumeSet = datanode.getDatanodeStateMachine().getContainer().getVolumeSet(); + return volumeSet.getFailedVolumesList().isEmpty() && volumeSet.getVolumesList().size() == DATA_VOLUMES; + } + + private static VolumeReplicationThreadPools volumePoolsOf(HddsDatanodeService dn) { + return dn.getDatanodeStateMachine().getSupervisor().getVolumeReplicationThreadPools(); + } + + /** + * The executor backing {@code volume}'s pool, so a test can prove a push ran on it via + * {@link ThreadPoolExecutor#getCompletedTaskCount()}. + */ + private static ThreadPoolExecutor volumePoolOf(HddsDatanodeService dn, HddsVolume volume) { + String volumeRoot = volume.getStorageDir().getPath(); + VolumeReplicationThreadPools pools = volumePoolsOf(dn); + assertThat(pools).isNotNull(); + assertThat(pools.hasPool(volumeRoot)).as("pool for %s", volumeRoot).isTrue(); + return (ThreadPoolExecutor) pools.getExecutor(volumeRoot); + } + + private static void assertVolumePools(HddsDatanodeService dn, int expectedVolumeCount, int expectedPoolSize) { + VolumeReplicationThreadPools pools = volumePoolsOf(dn); + assertThat(pools).isNotNull(); + List<? extends StorageVolume> volumes = dn.getDatanodeStateMachine().getContainer().getVolumeSet().getVolumesList(); + assertThat(volumes).hasSize(expectedVolumeCount); + for (StorageVolume volume : volumes) { + String volumeRoot = volume.getStorageDir().getPath(); + assertThat(pools.hasPool(volumeRoot)).as("pool for %s", volumeRoot).isTrue(); + assertThat(pools.getPoolSize(volumeRoot)).as("pool size for %s", volumeRoot).isEqualTo(expectedPoolSize); + } + } + + private static void waitForVolumePoolSize(HddsDatanodeService dn, int expectedPoolSize) + throws TimeoutException, InterruptedException { + List<? extends StorageVolume> volumes = dn.getDatanodeStateMachine().getContainer().getVolumeSet().getVolumesList(); + GenericTestUtils.waitFor(() -> { + VolumeReplicationThreadPools pools = volumePoolsOf(dn); + return pools != null && volumes.stream() + .allMatch(v -> pools.getPoolSize(v.getStorageDir().getPath()) == expectedPoolSize); + }, 200, 30000); + assertVolumePools(dn, DATA_VOLUMES, expectedPoolSize); + } + + private static void waitForVolumePoolState(HddsDatanodeService sourceDn, HddsVolume failedVolume, + HddsVolume healthyVolume) throws TimeoutException, InterruptedException { + String failedPath = failedVolume.getStorageDir().getPath(); + String healthyPath = healthyVolume.getStorageDir().getPath(); + GenericTestUtils.waitFor(() -> { + VolumeReplicationThreadPools pools = volumePoolsOf(sourceDn); + return pools != null && !pools.hasPool(failedPath) && pools.hasPool(healthyPath); + }, 100, 60000); + VolumeReplicationThreadPools pools = volumePoolsOf(sourceDn); + assertThat(pools).isNotNull(); + assertThat(pools.hasPool(failedPath)).as("pool for failed volume %s", failedPath).isFalse(); + assertThat(pools.hasPool(healthyPath)).as("pool for healthy volume %s", healthyPath).isTrue(); + } + + private static void queuePushAndWaitForContainer(MiniOzoneCluster cluster, ReplicateContainerCommand cmd, + DatanodeDetails source, DatanodeDetails target, long containerId) + throws IOException, InterruptedException, TimeoutException { + queueReplicationCommand(cluster, cmd, source); + GenericTestUtils.waitFor(() -> hasContainer(cluster, target, containerId), 100, 30000); + } + + private static void queuePushAndWaitForFailure(MiniOzoneCluster cluster, ReplicateContainerCommand cmd, + DatanodeDetails source, ReplicationSupervisor supervisor, long previousFailureCount) + throws IOException, InterruptedException, TimeoutException { + queueReplicationCommand(cluster, cmd, source); + GenericTestUtils.waitFor( + () -> supervisor.getReplicationFailureCount(ReplicationTask.METRIC_NAME) >= previousFailureCount + 1, + 100, 30000); + } + + private static void queueReplicationCommand(MiniOzoneCluster cluster, + ReplicateContainerCommand cmd, DatanodeDetails source) throws IOException { + DatanodeStateMachine stateMachine = cluster.getHddsDatanode(source).getDatanodeStateMachine(); + StateContext context = stateMachine.getContext(); + context.getTermOfLeaderSCM().ifPresent(cmd::setTerm); + context.addCommand(cmd); + } + + private static boolean hasContainer(MiniOzoneCluster cluster, + DatanodeDetails datanode, long containerId) { + try { + return cluster.getHddsDatanode(datanode).getDatanodeStateMachine().getContainer() + .getContainerSet().getContainer(containerId) != null; + } catch (IOException e) { + return false; + } + } + + private static long findOrCreateContainerOnVolume(MiniOzoneCluster cluster, + XceiverClientFactory clientFactory, DatanodeDetails dn, HddsVolume targetVolume) throws Exception { + for (int attempt = 0; attempt < 30; attempt++) { + long containerId = createClosedContainer(clientFactory, dn); + Container<?> container = getContainer(cluster, dn, containerId); + if (targetVolume.equals(container.getContainerData().getVolume())) { + return containerId; + } + } + throw new AssertionError("Could not place container on volume " + targetVolume); + } + + private static long createClosedContainer(XceiverClientFactory clientFactory, DatanodeDetails dn) + throws Exception { + long containerId = CONTAINER_ID.incrementAndGet(); + try (XceiverClientSpi client = clientFactory.acquireClient(createPipeline(singleton(dn)))) { + createContainer(client, containerId, null, CLOSED, 0); + } + return containerId; + } + + private static Container<?> getContainer(MiniOzoneCluster cluster, DatanodeDetails datanode, long containerId) + throws IOException { + HddsDatanodeService dnService = cluster.getHddsDatanode(datanode); + Container<?> container = dnService.getDatanodeStateMachine().getContainer() + .getContainerSet().getContainer(containerId); + if (container == null) { + throw new AssertionError("Container " + containerId + " not found on " + datanode); + } + return container; + } + + private static MiniOzoneCluster newCluster(OzoneConfiguration conf, int numDatanodes) throws IOException { + UniformDatanodesFactory uniformFactory = UniformDatanodesFactory.newBuilder() + .setNumDataVolumes(DATA_VOLUMES) + .build(); + return MiniOzoneCluster.newBuilder(conf) + .setNumDatanodes(numDatanodes) + .setDatanodeFactory(baseConf -> { + OzoneConfiguration dnConf = uniformFactory.apply(baseConf); + // UniformDatanodesFactory assigns the replication port with + // setFromObject(new ReplicationConfig().setPort(..)), which rewrites every field of that config + // back to its default and so drops the per-volume settings from the cluster conf. Re-apply them. + dnConf.setBoolean(ReplicationServer.ReplicationConfig.PER_VOLUME_ENABLED_KEY, true); + dnConf.setInt(ReplicationServer.ReplicationConfig.PER_VOLUME_STREAMS_LIMIT_KEY, PER_VOLUME_STREAMS); + return dnConf; + }) + .build(); + } + + private static OzoneConfiguration createConfig() { + OzoneConfiguration conf = new OzoneConfiguration(); + conf.setTimeDuration(OZONE_SCM_STALENODE_INTERVAL, 3, TimeUnit.SECONDS); + conf.setTimeDuration(OZONE_SCM_DEADNODE_INTERVAL, 6, TimeUnit.SECONDS); + conf.setTimeDuration(HDDS_HEARTBEAT_INTERVAL, 1, TimeUnit.SECONDS); + conf.setTimeDuration(HDDS_PIPELINE_REPORT_INTERVAL, 1, TimeUnit.SECONDS); + conf.setTimeDuration(HDDS_CONTAINER_REPORT_INTERVAL, 1, TimeUnit.SECONDS); + conf.setTimeDuration(HDDS_NODE_REPORT_INTERVAL, 1, TimeUnit.SECONDS); + conf.setTimeDuration(OZONE_SCM_HEARTBEAT_PROCESS_INTERVAL, 100, TimeUnit.MILLISECONDS); + conf.setTimeDuration(OZONE_SCM_DATANODE_ADMIN_MONITOR_INTERVAL, 1, TimeUnit.SECONDS); + conf.setTimeDuration(HddsConfigKeys.HDDS_SCM_WAIT_TIME_AFTER_SAFE_MODE_EXIT, 0, TimeUnit.SECONDS); + conf.setTimeDuration(ScmConfigKeys.OZONE_SCM_EXPIRED_CONTAINER_REPLICA_OP_SCRUB_INTERVAL, 1, TimeUnit.SECONDS); + conf.setStorageSize(ScmConfigKeys.OZONE_SCM_CONTAINER_SIZE, 5, StorageUnit.MB); + conf.setStorageSize(OzoneConfigKeys.OZONE_SCM_BLOCK_SIZE, 1, StorageUnit.MB); + conf.set(OzoneConfigKeys.OZONE_SCM_CLOSE_CONTAINER_WAIT_DURATION, "2s"); + conf.set(ScmConfigKeys.OZONE_SCM_PIPELINE_SCRUB_INTERVAL, "2s"); + conf.set(ScmConfigKeys.OZONE_SCM_PIPELINE_DESTROY_TIMEOUT, "5s"); + conf.setInt(ScmConfigKeys.OZONE_DATANODE_PIPELINE_LIMIT, 1); + + ReplicationManagerConfiguration repConf = conf.getObject(ReplicationManagerConfiguration.class); + repConf.setInterval(Duration.ofSeconds(1)); + repConf.setUnderReplicatedInterval(Duration.ofSeconds(1)); + repConf.setOverReplicatedInterval(Duration.ofSeconds(1)); + conf.setFromObject(repConf); + + ReplicationServer.ReplicationConfig replicationConfig = + conf.getObject(ReplicationServer.ReplicationConfig.class); + replicationConfig.setPerVolumeEnabled(true); + replicationConfig.setPerVolumeStreamsLimit(PER_VOLUME_STREAMS); + conf.setFromObject(replicationConfig); + + DatanodeConfiguration dnConf = conf.getObject(DatanodeConfiguration.class); + dnConf.setFailedDataVolumesTolerated(1); + dnConf.setDiskCheckMinGap(Duration.ofSeconds(0)); + dnConf.setPeriodicDiskCheckIntervalMinutes(1); + conf.setFromObject(dnConf); + return conf; + } + + private static void triggerAndWaitForVolumeFailure(MutableVolumeSet volSet, StorageVolume volume) + throws Exception { + DatanodeTestUtils.simulateBadVolume(volume); + volSet.checkVolumeAsync(volume); + DatanodeTestUtils.waitForHandleFailedVolume(volSet, 1); + } + + private static void generateData(OzoneBucket bucket, int keyCount, String keyPrefix, + ReplicationConfig replicationConfig) throws IOException { + for (int i = 0; i < keyCount; i++) { + DataTestUtil.createKey(bucket, keyPrefix + i, replicationConfig, "this is the content".getBytes(UTF_8)); + } + } + + private static ContainerInfo waitForKeyContainer(OzoneBucket bucket, + ContainerManager cm, String keyName, int expectedReplicas) throws Exception { + OzoneKeyDetails keyDetails = bucket.getKey(keyName); + long containerId = keyDetails.getOzoneKeyLocations().get(0).getContainerID(); + ContainerInfo container = cm.getContainer(ContainerID.valueOf(containerId)); + waitForContainerReplicas(cm, container, expectedReplicas); + return container; + } + + private static void waitForContainerReplicas(ContainerManager cm, ContainerInfo container, int count) + throws TimeoutException, InterruptedException { + waitForReplicas(cm, container.containerID(), count); + } + + /** + * Waits until SCM reports exactly {@code count} replicas. Kept at 30s per wait so that the eight waits in + * {@link #testDecommissionWithPerVolumePools} stay inside the 5m default JUnit timeout the root pom sets, + * and a genuine stall is reported by the predicate rather than by an anonymous test timeout. + */ + private static void waitForReplicas(ContainerManager cm, ContainerID containerId, int count) + throws TimeoutException, InterruptedException { + GenericTestUtils.waitFor(() -> replicasOf(cm, containerId).size() == count, 200, 30000); + } + + /** SCM's replica set, treating a transient lookup failure as "not ready yet". */ + private static Set<ContainerReplica> replicasOf(ContainerManager cm, ContainerID containerId) { + try { + return cm.getContainerReplicas(containerId); + } catch (Exception e) { + return emptySet(); + } + } + + } diff --cc hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOMUpgradeFinalization.java index 0bc4826820e,be0f9909977..bac87cbf6b3 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOMUpgradeFinalization.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOMUpgradeFinalization.java @@@ -230,39 -72,50 +230,39 @@@ class TestOMUpgradeFinalization cluster.waitForClusterToBeReady(); try (OzoneClient client = cluster.newClient()) { - List<OzoneManager> runningOms = cluster.getOzoneManagersList(); - final int shutdownOMIndex = 2; - OzoneManager downedOM = cluster.getOzoneManager(shutdownOMIndex); - cluster.stopOzoneManager(shutdownOMIndex); - assertFalse(downedOM.isRunning()); - assertEquals(runningOms.remove(shutdownOMIndex), downedOM); - - OzoneManagerProtocol omClient = client.getObjectStore().getClientProxy() - .getOzoneManagerClient(); - // Have to do a "prepare" operation to get rid of the logs in the active - // OMs. - long prepareIndex = omClient.prepareOzoneManager(120L, 5L); - assertClusterPrepared(prepareIndex, runningOms); - AuditLogTestUtils.verifyAuditLog(OMAction.UPGRADE_PREPARE, AuditEventStatus.SUCCESS); - - omClient.cancelOzoneManagerPrepare(); - AuditLogTestUtils.verifyAuditLog(OMAction.UPGRADE_CANCEL, AuditEventStatus.SUCCESS); - StatusAndMessages response = - omClient.finalizeUpgrade("finalize-test"); - System.out.println("Finalization Messages : " + response.msgs()); - AuditLogTestUtils.verifyAuditLog(OMAction.UPGRADE_FINALIZE, AuditEventStatus.SUCCESS); - - waitForFinalization(omClient); - cluster.restartOzoneManager(downedOM, true); - - OzoneManagerStateMachine omStateMachine = downedOM.getOmRatisServer() - .getOmStateMachine(); - try { - waitFor(() -> omStateMachine.getLifeCycleState().isPausingOrPaused(), - 1000, 60000); - } catch (TimeoutException timeEx) { - assertEquals(LifeCycle.State.RUNNING, - omStateMachine.getLifeCycle().getCurrentState()); + OzoneManagerProtocol omClient = client.getObjectStore().getClientProxy().getOzoneManagerClient(); + + // Every OM is pre-finalized for ZDU. The version returned to clients should reflect the last software + // version before ZDU to remain compatible with the old version framework. - assertAllOmServiceInfoVersions(cluster, OzoneManagerVersion.GET_FILE_STATUS_REJECTS_OBS); ++ assertAllOmServiceInfoVersions(cluster, OzoneManagerVersion.LOOKUP_FILE_REJECTS_OBS); + + omClient.finalizeUpgrade(); + OMUpgradeTestUtils.waitForFinalization(omClient); + for (OzoneManager om : cluster.getOzoneManagersList()) { + waitFor(() -> !om.getVersionManager().needsFinalization(), 100, 30000); } - waitFor(() -> !omStateMachine.getLifeCycle().getCurrentState() - .isPausingOrPaused(), 1000, 60000); + // After finalization: every OM advertises the software version for all three OMs. + assertAllOmServiceInfoVersions(cluster, OzoneManagerVersion.SOFTWARE_VERSION); + } + } + } + + /** + * Assert that, from every OM's {@code getServiceList()}, all three OM entries report {@code expected}. + */ + private static void assertAllOmServiceInfoVersions(MiniOzoneHAClusterImpl cluster, OzoneManagerVersion expected) + throws IOException { + for (OzoneManager om : cluster.getOzoneManagersList()) { + assertEquals(expected, om.getVersionManager().getVersionForClient()); - assertEquals(maxLayoutVersion(), - downedOM.getVersionManager().getMetadataLayoutVersion()); - String lvString = downedOM.getMetadataManager().getMetaTable() - .get(LAYOUT_VERSION_KEY); - assertNotNull(lvString); - assertEquals(maxLayoutVersion(), Integer.parseInt(lvString)); + List<OzoneManagerVersion> omVersions = om.getServiceList().stream() + .filter(info -> info.getNodeType() == HddsProtos.NodeType.OM) + .map(info -> OzoneManagerVersion.deserialize(info.getProtobuf().getOMVersion())) + .collect(Collectors.toList()); + assertEquals(3, omVersions.size()); + for (OzoneManagerVersion version : omVersions) { + assertEquals(expected, version); } } } diff --cc hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/file/OMFileCreateRequestWithFSO.java index b47ebd46948,dea0782013f..0842a5b1aae --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/file/OMFileCreateRequestWithFSO.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/file/OMFileCreateRequestWithFSO.java @@@ -240,7 -134,204 +135,204 @@@ public class OMFileCreateRequestWithFS } } - // Audit Log outside the lock + auditAndLogResult(ozoneManager, createFileRequest, auditMap, exception, result, numKeysCreated); + + return omClientResponse; + } + + /** + * Phase 1 (no bucket lock): resolves the path and prepares the open-file entry, the missing parent + * directories and the quota delta. All of this reads committed state only. The OM apply path is + * single-threaded (OzoneManagerStateMachine uses a single-thread executor), so no other transaction + * can change what these reads observe before Phase 2 mutates; the double-buffer flush/cleanup + * threads only materialize already-committed epochs and never alter a key's visible value. + * <p> + * Keeping this walk out of the bucket write lock is the point of HDDS-16289: it stops the lone apply + * thread from gating readers of a hot bucket (getBucketInfo/getFileStatus/lookupKey) during the + * per-segment path resolution. This rests on the serial-apply invariant: if OM ever applies + * transactions in parallel per bucket/key, the re-check in {@link #applyFileCreate} is not enough and + * every read here needs re-validating under the lock, the walked ancestors and the missing-parent set + * as well as the leaf. + */ + private PreparedFileCreate prepareFileCreate(OzoneManager ozoneManager, + CreateFileRequest createFileRequest, long trxnLogIndex) throws IOException { + OMMetadataManager omMetadataManager = ozoneManager.getMetadataManager(); + KeyArgs keyArgs = createFileRequest.getKeyArgs(); + String volumeName = keyArgs.getVolumeName(); + String bucketName = keyArgs.getBucketName(); + String keyName = keyArgs.getKeyName(); + // if isOverWrite is true, file would be over written. + boolean isOverWrite = createFileRequest.getIsOverwrite(); + + validateBucketAndVolume(omMetadataManager, volumeName, bucketName); + + final long volumeId = omMetadataManager.getVolumeId(volumeName); + final long bucketId = omMetadataManager + .getBucketId(volumeName, bucketName); + + OmKeyInfo dbFileInfo = null; + + OMFileRequest.OMPathInfoWithFSO pathInfoFSO = + OMFileRequest.verifyDirectoryKeysInPath(omMetadataManager, + volumeName, bucketName, keyName, Paths.get(keyName)); + + final String dbFileKey = omMetadataManager.getOzonePathKey(volumeId, + bucketId, pathInfoFSO.getLastKnownParentId(), + pathInfoFSO.getLeafNodeName()); + if (pathInfoFSO.getDirectoryResult() + == OMFileRequest.OMDirectoryResult.FILE_EXISTS) { + dbFileInfo = OMFileRequest.getOmKeyInfoFromFileTable(false, + omMetadataManager, dbFileKey, keyName); + } + + // check if the file or directory already existed in OM + checkDirectoryResult(keyName, isOverWrite, + pathInfoFSO.getDirectoryResult()); + + // if isRecursive is true, file would be created even if parent + // directories does not exist. + if (!createFileRequest.getIsRecursive()) { + checkAllParentsExist(keyArgs, pathInfoFSO); + } + + // do open key + // Read-only copy, used here only to inherit ACLs, encryption and replication defaults. The quota + // read-modify-publish in Phase 2 takes its own getBucketInfoForUpdate copy under the lock. + OmBucketInfo bucketInfo = omMetadataManager.getBucketTable().get( + omMetadataManager.getBucketKey(volumeName, bucketName)); + // add all missing parents to dir table + + List<OmDirectoryInfo> missingParentInfos = getAllMissingParentDirInfo( + ozoneManager, keyArgs, bucketInfo, pathInfoFSO, trxnLogIndex); + + final ReplicationConfig repConfig = OzoneConfigUtil + .resolveReplicationConfigPreference(keyArgs.getType(), + keyArgs.getFactor(), keyArgs.getEcReplicationConfig(), + bucketInfo.getDefaultReplicationConfig(), + ozoneManager); + + OmKeyInfo omFileInfo = prepareFileInfo(omMetadataManager, keyArgs, + dbFileInfo, keyArgs.getDataSize(), new ArrayList<>(), + getFileEncryptionInfo(keyArgs), ozoneManager.getPrefixManager(), + bucketInfo, pathInfoFSO, trxnLogIndex, + pathInfoFSO.getLeafNodeObjectId(), + repConfig, ozoneManager.getConfig()); + validateEncryptionKeyInfo(bucketInfo, keyArgs); + + String dbOpenFileName = omMetadataManager + .getOpenFileName(volumeId, bucketId, + pathInfoFSO.getLastKnownParentId(), + pathInfoFSO.getLeafNodeName(), createFileRequest.getClientID()); + + // Key of the leaf in the file table, for the Phase 2 re-check. Built from the parent id + // getAllMissingParentDirInfo left on pathInfoFSO, so it addresses the leaf even when intermediate + // directories were missing and this transaction creates them. dbFileKey above predates that call + // and can still name the deepest pre-existing ancestor, which is a different directory. + final String dbLeafFileKey = omMetadataManager.getOzonePathKey(volumeId, bucketId, + pathInfoFSO.getLastKnownParentId(), pathInfoFSO.getLeafNodeName()); + + // Append new blocks + List<OmKeyLocationInfo> newLocationList = keyArgs.getKeyLocationsList() + .stream().map(OmKeyLocationInfo::getFromProtobuf) + .collect(Collectors.toList()); + omFileInfo.appendNewBlocks(newLocationList, false); + + final long preAllocatedSpace = + newLocationList.size() * ozoneManager.getScmBlockSize() * repConfig + .getRequiredNodes(); + + return new PreparedFileCreate(volumeId, bucketId, dbLeafFileKey, omFileInfo, + missingParentInfos, dbOpenFileName, preAllocatedSpace); + } + + /** + * Phase 2 (under the bucket write lock): re-checks the overwrite guard resolved in Phase 1, then + * applies the open-file and directory cache entries and the bucket quota charge and publishes the + * bucket copy. + */ + private OMClientResponse applyFileCreate(OMMetadataManager omMetadataManager, + CreateFileRequest createFileRequest, PreparedFileCreate prepared, long trxnLogIndex, + OMResponse.Builder omResponse) throws IOException { + KeyArgs keyArgs = createFileRequest.getKeyArgs(); + String volumeName = keyArgs.getVolumeName(); + String bucketName = keyArgs.getBucketName(); + String keyName = keyArgs.getKeyName(); + + // Cheap O(1) re-check of the overwrite guard resolved in Phase 1. Under serial apply this always + // holds, so it is unreachable; it is kept because it is a point lookup and it fails the way + // checkDirectoryResult would rather than silently overwriting. It does not make Phase 1 safe under + // concurrent apply: it re-reads the leaf, not the ancestor chain lastKnownParentId came from. + if (!createFileRequest.getIsOverwrite() + && omMetadataManager.getKeyTable(getBucketLayout()).isExist(prepared.dbLeafFileKey)) { + throw new OMException("File " + keyName + " already exists", + OMException.ResultCodes.FILE_ALREADY_EXISTS); + } + + OmBucketInfo omBucketInfo = getBucketInfoForUpdate(omMetadataManager, volumeName, bucketName); + // check bucket and volume quota + checkBucketQuotaInBytes(omMetadataManager, omBucketInfo, + prepared.preAllocatedSpace); + final int numKeysCreated = prepared.missingParentInfos.size(); + checkBucketQuotaInNamespace(omBucketInfo, numKeysCreated + 1L); + omBucketInfo.incrUsedNamespace(numKeysCreated); + + OMFileRequest.addOpenFileTableCacheEntry(omMetadataManager, + prepared.dbOpenFileName, prepared.omFileInfo, keyName, trxnLogIndex); + + // Add cache entries for the prefix directories. + // Skip adding for the file key itself, until Key Commit. + OMFileRequest.addDirectoryTableCacheEntries(omMetadataManager, prepared.volumeId, + prepared.bucketId, trxnLogIndex, prepared.missingParentInfos, null); + + omMetadataManager.getBucketTable().addCacheEntry( + omMetadataManager.getBucketKey(volumeName, bucketName), omBucketInfo, trxnLogIndex); + + // Prepare response. Sets user given full key name in the 'keyName' + // attribute in response object. - int clientVersion = getOmRequest().getVersion(); ++ ClientVersion clientVersion = ClientVersion.deserialize(getOmRequest().getVersion()); + long openVersion = prepared.omFileInfo.getLatestVersionLocations().getVersion(); + omResponse.setCreateFileResponse(CreateFileResponse.newBuilder() + .setKeyInfo(prepared.omFileInfo.getNetworkProtobuf(keyName, clientVersion, + keyArgs.getLatestVersionLocation())) + .setID(createFileRequest.getClientID()) + .setOpenVersion(openVersion).build()) + .setCmdType(Type.CreateFile); + return new OMFileCreateResponseWithFSO(omResponse.build(), + prepared.omFileInfo, prepared.missingParentInfos, createFileRequest.getClientID(), + omBucketInfo.copyObject(), prepared.volumeId); + } + + /** + * Phase 1 output of {@link #prepareFileCreate}, consumed by {@link #applyFileCreate} under the + * bucket write lock: the resolved path keys, the open file and missing parent directories to cache, + * and the quota deltas to charge. + */ + private static final class PreparedFileCreate { + private final long volumeId; + private final long bucketId; + private final String dbLeafFileKey; + private final OmKeyInfo omFileInfo; + private final List<OmDirectoryInfo> missingParentInfos; + private final String dbOpenFileName; + private final long preAllocatedSpace; + + PreparedFileCreate(long volumeId, long bucketId, String dbLeafFileKey, OmKeyInfo omFileInfo, + List<OmDirectoryInfo> missingParentInfos, String dbOpenFileName, long preAllocatedSpace) { + this.volumeId = volumeId; + this.bucketId = bucketId; + this.dbLeafFileKey = dbLeafFileKey; + this.omFileInfo = omFileInfo; + this.missingParentInfos = missingParentInfos; + this.dbOpenFileName = dbOpenFileName; + this.preAllocatedSpace = preAllocatedSpace; + } + } + + /** + * Emits the audit log and the result log outside the bucket lock. + */ + private void auditAndLogResult(OzoneManager ozoneManager, CreateFileRequest createFileRequest, + Map<String, String> auditMap, Exception exception, Result result, int numKeysCreated) { markForAudit(ozoneManager.getAuditLogger(), buildAuditMessage( OMAction.CREATE_FILE, auditMap, exception, getOmRequest().getUserInfo())); diff --cc hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/upgrade/TestOMVersionManager.java index 7a10b57dcd5,388b5134c41..f7f727cacd4 --- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/upgrade/TestOMVersionManager.java +++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/upgrade/TestOMVersionManager.java @@@ -226,29 -107,38 +226,29 @@@ class TestOMVersionManager extends Abst } @Test - public void testOmUpgradeActionsRegistered() throws Exception { - OMLayoutVersionManager lvm = new OMLayoutVersionManager(); // MLV >= 0 - assertFalse(lvm.needsFinalization()); - - // INITIAL_VERSION is finalized, hence should not register. - Optional<OmUpgradeAction> action = - INITIAL_VERSION.action(); - assertFalse(action.isPresent()); - - lvm = mock(OMLayoutVersionManager.class); - when(lvm.getMetadataLayoutVersion()).thenReturn(-1); - doCallRealMethod().when(lvm).registerUpgradeActions(anyString()); - lvm.registerUpgradeActions(OM_UPGRADE_CLASS_PACKAGE); - - action = INITIAL_VERSION.action(); - assertTrue(action.isPresent()); - assertEquals(MockOmUpgradeAction.class, action.get().getClass()); - OzoneManager omMock = mock(OzoneManager.class); - action.get().execute(omMock); - verify(omMock, times(1)).getVersion(); + public void testGetVersionForClientWhenPreZdu() throws Exception { + // A pre-ZDU apparent version clamps to the last pre-ZDU client version. + try (OMVersionManager versionManager = createManager(INITIAL_VERSION.serialize())) { - assertEquals(OzoneManagerVersion.GET_FILE_STATUS_REJECTS_OBS, versionManager.getVersionForClient()); ++ assertEquals(OzoneManagerVersion.LOOKUP_FILE_REJECTS_OBS, versionManager.getVersionForClient()); + } } - /** - * Mock OM upgrade action class. - */ - @UpgradeActionOm(feature = - INITIAL_VERSION) - public static class MockOmUpgradeAction implements OmUpgradeAction { + private OMStorage newOmStorage(int apparentVersion) + throws IOException { + // Reinitialize the configuration to point to a new unique storage location. + Path dbDir = Files.createDirectory(new File(tempFolder.toFile(), UUID.randomUUID().toString()).toPath()); + conf.set(OMConfigKeys.OZONE_OM_DB_DIRS, dbDir.toString()); + OMStorage storage = new OMStorage(conf); + storage.setClusterId("test-cluster"); + storage.setApparentVersion(apparentVersion); + storage.setOmId(UUID.randomUUID().toString()); + storage.initialize(); + storage.persistCurrentState(); + return storage; + } - @Override - public void execute(OzoneManager arg) { - arg.getVersion(); - } + private static void assertOmApparentVersionOnDisk(OzoneConfiguration conf, int expected) throws IOException { + OMStorage reloaded = new OMStorage(conf); + assertEquals(expected, reloaded.getApparentVersion()); } } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
