You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@ozone.apache.org by "jojochuang (via GitHub)" <gi...@apache.org> on 2026/08/28 17:57:21 UTC
[PR] HDDS-15802. Add integration tests for per-volume push replication [ozone]
jojochuang opened a new pull request, #11146:
URL: https://github.com/apache/ozone/pull/11146
## Summary
Revives closed #10703 as a **tests-only** follow-up to HDDS-15412 (#10693).
Adds `TestPerVolumePushReplication` with three P0 integration scenarios on a shared mini-cluster:
- Push and SCM-driven under-replication recovery with per-volume pools enabled
- Volume failure: failed-volume pool removed, healthy-volume push replication continues
- Decommission E2E with Ratis and EC data while per-volume pools are enabled
## Jira
https://issues.apache.org/jira/browse/HDDS-15802
## Notes
- No production code changes; feature landed in #10693.
- Failed-volume push tasks use global pool fallback (HDDS-15327); replication fails on the bad volume in the integration test.
## Test plan
- [x] `mvn -pl :ozone-integration-test test -Dtest=TestPerVolumePushReplication -DskipShade -DskipRecon -DskipDocs`
Generated-by: Cursor (Auto)
Made with [Cursor](https://cursor.com)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-15802. Add integration tests for per-volume push replication [ozone]
Posted by "smengcl (via GitHub)" <gi...@apache.org>.
smengcl commented on code in PR #11146:
URL: https://github.com/apache/ozone/pull/11146#discussion_r4100788276
##########
hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/replication/TestPerVolumePushReplication.java:
##########
@@ -0,0 +1,547 @@
+/*
+ * 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.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);
+ 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);
+ 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);
+ 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);
+ }
Review Comment:
Correct, and thanks for pushing this further than I did. I had noted that `restoreBadVolume` only flips the directory permission but stopped at "it does not undo as much as the name suggests"; you named the actual failure. `assertVolumePools` asserts `getVolumesList()` has `DATA_VOLUMES` entries, the failed volume is gone from that list for good, so the decommission test fails whenever the Ratis and EC pipeline intersection lands on this datanode.
Fixed in f5fb92ae38c by restarting the datanode in the same `finally` block, right after the permission is restored, which is also what the first test already does for the node it stops.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-15802. Add integration tests for per-volume push replication [ozone]
Posted by "smengcl (via GitHub)" <gi...@apache.org>.
smengcl commented on code in PR #11146:
URL: https://github.com/apache/ozone/pull/11146#discussion_r4100707068
##########
hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/replication/TestPerVolumePushReplication.java:
##########
@@ -0,0 +1,545 @@
+/*
+ * 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.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.OzoneTestHelper.waitForReplicaCount;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.BooleanSupplier;
+import java.util.stream.Collectors;
+import org.apache.hadoop.hdds.HddsConfigKeys;
+import org.apache.hadoop.hdds.client.BlockID;
+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.protocol.datanode.proto.ContainerProtos;
+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.ContainerTestHelper;
+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;
+import org.junit.jupiter.api.parallel.Execution;
+import org.junit.jupiter.api.parallel.ExecutionMode;
+import org.junit.jupiter.api.parallel.ResourceLock;
+
+/**
+ * Integration tests for per-volume push replication thread pools (HDDS-15412).
+ */
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
+@Execution(ExecutionMode.SAME_THREAD)
+@ResourceLock("MiniOzoneCluster")
+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 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 OzoneBucket bucket;
+
+ @BeforeAll
+ void setUp() throws Exception {
+ OzoneConfiguration conf = createSharedConfig();
+ cluster = newCluster(conf, DATANODE_COUNT);
+ cluster.waitForClusterToBeReady();
+ clientFactory = new XceiverClientManager(conf);
+ try (OzoneClient client = cluster.newClient()) {
+ bucket = DataTestUtil.createVolumeAndBucket(client, VOLUME, BUCKET);
+ }
+ }
+
+ @AfterAll
+ void tearDown() {
+ IOUtils.closeQuietly(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, 0L);
+
+ ReplicateContainerCommand cmd =
+ ReplicateContainerCommand.toTarget(containerId, target);
+ queuePushAndWaitForContainer(cluster, cmd, source, target, containerId);
+ getContainer(cluster, target, containerId);
+ assertVolumePools(sourceDn, DATA_VOLUMES, 1);
+
+ 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);
+
+ ContainerManager containerManager = cluster.getStorageContainerManager().getContainerManager();
+ Set<ContainerReplica> replicas =
+ containerManager.getContainerReplicas(ContainerID.valueOf(scmContainerId));
+ DatanodeDetails replicaDn = replicas.iterator().next().getDatanodeDetails();
+ cluster.shutdownHddsDatanode(replicaDn);
+ waitForReplicaCount(scmContainerId, 3, cluster);
+ }
+
+ @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, 0L);
+ long containerOnVol1 = findOrCreateContainerOnVolume(
+ cluster, clientFactory, source, vol1, 0L);
+
+ assertVolumePools(sourceDn, DATA_VOLUMES, 1);
+
+ triggerAndWaitForVolumeFailure(volSet, vol0);
+ waitForVolumePoolState(sourceDn, vol0, vol1);
+
+ ReplicateContainerCommand cmd =
+ ReplicateContainerCommand.toTarget(containerOnVol1, target);
+ queuePushAndWaitForContainer(cluster, cmd, source, target, containerOnVol1);
+ getContainer(cluster, target, containerOnVol1);
+ assertEquals(1, volSet.getFailedVolumesList().size());
+
+ // Task routes via global pool fallback (HDDS-15327); replication fails on bad volume.
+ ReplicateContainerCommand failedVolCmd =
+ ReplicateContainerCommand.toTarget(containerOnVol0, target);
+ ReplicationSupervisor supervisor =
+ sourceDn.getDatanodeStateMachine().getSupervisor();
+ long previousFailures = supervisor.getReplicationFailureCount();
+ queuePushAndWaitForFailure(cluster, failedVolCmd, source, target,
+ containerOnVol0, supervisor, previousFailures);
+ assertTrue(supervisor.getReplicationFailureCount() >= previousFailures + 1);
+
+ DatanodeTestUtils.restoreBadVolume(vol0);
+ }
+
+ @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);
+
+ scmClient.decommissionNodes(singletonList(getDNHostAndPort(toDecommission)), false);
+ waitForDnToReachOpState(nm, toDecommission, DECOMMISSIONED);
+
+ 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(StandardCharsets.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) {
+ 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 void assertVolumePools(HddsDatanodeService dn,
+ int expectedVolumeCount, int expectedPoolSize) {
+ VolumeReplicationThreadPools pools =
+ dn.getDatanodeStateMachine().getSupervisor().getVolumeReplicationThreadPools();
+ assertNotNull(pools);
+ List<? extends StorageVolume> volumes =
+ dn.getDatanodeStateMachine().getContainer().getVolumeSet().getVolumesList();
+ assertEquals(expectedVolumeCount, volumes.size());
+ for (StorageVolume volume : volumes) {
+ String volumeRoot = volume.getStorageDir().getPath();
+ assertTrue(pools.hasPool(volumeRoot), "missing pool for " + volumeRoot);
+ assertEquals(expectedPoolSize, pools.getPoolSize(volumeRoot));
+ }
+ }
+
+ 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((BooleanSupplier) () -> {
+ VolumeReplicationThreadPools pools =
+ sourceDn.getDatanodeStateMachine().getSupervisor().getVolumeReplicationThreadPools();
+ return pools != null
+ && !pools.hasPool(failedPath)
+ && pools.hasPool(healthyPath);
+ }, 100, 60000);
+ VolumeReplicationThreadPools pools =
+ sourceDn.getDatanodeStateMachine().getSupervisor().getVolumeReplicationThreadPools();
+ assertNotNull(pools);
+ assertFalse(pools.hasPool(failedPath));
+ assertTrue(pools.hasPool(healthyPath));
+ }
+
+ private static void queuePushAndWaitForContainer(MiniOzoneCluster cluster,
+ ReplicateContainerCommand cmd, DatanodeDetails source,
+ DatanodeDetails target, long containerId)
+ throws IOException, InterruptedException, TimeoutException {
+ queueReplicationCommand(cluster, cmd, source);
+ GenericTestUtils.waitFor((BooleanSupplier) () ->
+ hasContainer(cluster, target, containerId), 100, 30000);
+ }
+
+ private static void queuePushAndWaitForFailure(MiniOzoneCluster cluster,
+ ReplicateContainerCommand cmd, DatanodeDetails source,
+ DatanodeDetails target, long containerId, ReplicationSupervisor supervisor,
+ long previousFailureCount)
+ throws IOException, InterruptedException, TimeoutException {
+ queueReplicationCommand(cluster, cmd, source);
+ GenericTestUtils.waitFor((BooleanSupplier) () ->
+ supervisor.getReplicationFailureCount() >= previousFailureCount + 1
+ && !hasContainer(cluster, target, containerId),
+ 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,
+ long dataSize) throws Exception {
+ for (int attempt = 0; attempt < 30; attempt++) {
+ long containerId = createClosedContainer(clientFactory, dn, dataSize);
+ 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, long dataSize) throws Exception {
+ long containerId = CONTAINER_ID.incrementAndGet();
+ try (XceiverClientSpi client = clientFactory.acquireClient(createPipeline(singleton(dn)))) {
+ if (dataSize <= 0) {
+ createContainer(client, containerId, null, CLOSED, 0);
+ return containerId;
+ }
+ createContainer(client, containerId, null);
Review Comment:
Addressed in 66bf80e1221. The `dataSize` parameter and the dead branch are gone, so `createClosedContainer` is now create, close, return the id.
##########
hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/replication/TestPerVolumePushReplication.java:
##########
@@ -0,0 +1,545 @@
+/*
+ * 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.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.OzoneTestHelper.waitForReplicaCount;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.BooleanSupplier;
+import java.util.stream.Collectors;
+import org.apache.hadoop.hdds.HddsConfigKeys;
+import org.apache.hadoop.hdds.client.BlockID;
+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.protocol.datanode.proto.ContainerProtos;
+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.ContainerTestHelper;
+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;
+import org.junit.jupiter.api.parallel.Execution;
+import org.junit.jupiter.api.parallel.ExecutionMode;
+import org.junit.jupiter.api.parallel.ResourceLock;
+
+/**
+ * Integration tests for per-volume push replication thread pools (HDDS-15412).
+ */
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
+@Execution(ExecutionMode.SAME_THREAD)
+@ResourceLock("MiniOzoneCluster")
+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 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 OzoneBucket bucket;
+
+ @BeforeAll
+ void setUp() throws Exception {
+ OzoneConfiguration conf = createSharedConfig();
+ cluster = newCluster(conf, DATANODE_COUNT);
+ cluster.waitForClusterToBeReady();
+ clientFactory = new XceiverClientManager(conf);
+ try (OzoneClient client = cluster.newClient()) {
+ bucket = DataTestUtil.createVolumeAndBucket(client, VOLUME, BUCKET);
+ }
+ }
+
+ @AfterAll
+ void tearDown() {
+ IOUtils.closeQuietly(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, 0L);
+
+ ReplicateContainerCommand cmd =
+ ReplicateContainerCommand.toTarget(containerId, target);
+ queuePushAndWaitForContainer(cluster, cmd, source, target, containerId);
+ getContainer(cluster, target, containerId);
+ assertVolumePools(sourceDn, DATA_VOLUMES, 1);
+
+ 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);
+
+ ContainerManager containerManager = cluster.getStorageContainerManager().getContainerManager();
+ Set<ContainerReplica> replicas =
+ containerManager.getContainerReplicas(ContainerID.valueOf(scmContainerId));
+ DatanodeDetails replicaDn = replicas.iterator().next().getDatanodeDetails();
+ cluster.shutdownHddsDatanode(replicaDn);
+ waitForReplicaCount(scmContainerId, 3, cluster);
+ }
+
+ @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, 0L);
+ long containerOnVol1 = findOrCreateContainerOnVolume(
+ cluster, clientFactory, source, vol1, 0L);
+
+ assertVolumePools(sourceDn, DATA_VOLUMES, 1);
+
+ triggerAndWaitForVolumeFailure(volSet, vol0);
+ waitForVolumePoolState(sourceDn, vol0, vol1);
+
+ ReplicateContainerCommand cmd =
+ ReplicateContainerCommand.toTarget(containerOnVol1, target);
+ queuePushAndWaitForContainer(cluster, cmd, source, target, containerOnVol1);
+ getContainer(cluster, target, containerOnVol1);
+ assertEquals(1, volSet.getFailedVolumesList().size());
+
+ // Task routes via global pool fallback (HDDS-15327); replication fails on bad volume.
+ ReplicateContainerCommand failedVolCmd =
+ ReplicateContainerCommand.toTarget(containerOnVol0, target);
+ ReplicationSupervisor supervisor =
+ sourceDn.getDatanodeStateMachine().getSupervisor();
+ long previousFailures = supervisor.getReplicationFailureCount();
+ queuePushAndWaitForFailure(cluster, failedVolCmd, source, target,
+ containerOnVol0, supervisor, previousFailures);
+ assertTrue(supervisor.getReplicationFailureCount() >= previousFailures + 1);
+
+ DatanodeTestUtils.restoreBadVolume(vol0);
+ }
+
+ @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);
+
+ scmClient.decommissionNodes(singletonList(getDNHostAndPort(toDecommission)), false);
+ waitForDnToReachOpState(nm, toDecommission, DECOMMISSIONED);
+
+ 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(StandardCharsets.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) {
+ 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 void assertVolumePools(HddsDatanodeService dn,
+ int expectedVolumeCount, int expectedPoolSize) {
+ VolumeReplicationThreadPools pools =
+ dn.getDatanodeStateMachine().getSupervisor().getVolumeReplicationThreadPools();
+ assertNotNull(pools);
+ List<? extends StorageVolume> volumes =
+ dn.getDatanodeStateMachine().getContainer().getVolumeSet().getVolumesList();
+ assertEquals(expectedVolumeCount, volumes.size());
+ for (StorageVolume volume : volumes) {
+ String volumeRoot = volume.getStorageDir().getPath();
+ assertTrue(pools.hasPool(volumeRoot), "missing pool for " + volumeRoot);
+ assertEquals(expectedPoolSize, pools.getPoolSize(volumeRoot));
+ }
+ }
+
+ 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((BooleanSupplier) () -> {
+ VolumeReplicationThreadPools pools =
+ sourceDn.getDatanodeStateMachine().getSupervisor().getVolumeReplicationThreadPools();
+ return pools != null
+ && !pools.hasPool(failedPath)
+ && pools.hasPool(healthyPath);
+ }, 100, 60000);
+ VolumeReplicationThreadPools pools =
+ sourceDn.getDatanodeStateMachine().getSupervisor().getVolumeReplicationThreadPools();
+ assertNotNull(pools);
+ assertFalse(pools.hasPool(failedPath));
+ assertTrue(pools.hasPool(healthyPath));
+ }
+
+ private static void queuePushAndWaitForContainer(MiniOzoneCluster cluster,
+ ReplicateContainerCommand cmd, DatanodeDetails source,
+ DatanodeDetails target, long containerId)
+ throws IOException, InterruptedException, TimeoutException {
+ queueReplicationCommand(cluster, cmd, source);
+ GenericTestUtils.waitFor((BooleanSupplier) () ->
+ hasContainer(cluster, target, containerId), 100, 30000);
+ }
+
+ private static void queuePushAndWaitForFailure(MiniOzoneCluster cluster,
+ ReplicateContainerCommand cmd, DatanodeDetails source,
+ DatanodeDetails target, long containerId, ReplicationSupervisor supervisor,
+ long previousFailureCount)
+ throws IOException, InterruptedException, TimeoutException {
+ queueReplicationCommand(cluster, cmd, source);
+ GenericTestUtils.waitFor((BooleanSupplier) () ->
+ supervisor.getReplicationFailureCount() >= previousFailureCount + 1
+ && !hasContainer(cluster, target, containerId),
+ 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);
Review Comment:
Not addressed in 66bf80e1221, deliberately, so leaving this thread open to track it.
Reusing `TestContainerReplication#queueAndWaitForCompletion`, `CONTAINER_ID` and `createNewClosedContainer` means widening their visibility and lifting them somewhere shared such as `OzoneTestHelper`, which edits a test class this PR does not otherwise touch. The duplication is real and worth removing, but I would rather do it as a follow up where the shared move is reviewable on its own instead of mixed into a new test class. Happy to do it here instead if you prefer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-15802. Add integration tests for per-volume push replication [ozone]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #11146:
URL: https://github.com/apache/ozone/pull/11146#issuecomment-5746299858
This PR has been marked as stale due to 21 days of inactivity. Please comment or remove the stale label to keep it open. Otherwise, it will be automatically closed in 7 days.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-15802. Add integration tests for per-volume push replication [ozone]
Posted by "rich7420 (via GitHub)" <gi...@apache.org>.
rich7420 commented on code in PR #11146:
URL: https://github.com/apache/ozone/pull/11146#discussion_r3886088698
##########
hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/replication/TestPerVolumePushReplication.java:
##########
@@ -0,0 +1,545 @@
+/*
+ * 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.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.OzoneTestHelper.waitForReplicaCount;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.BooleanSupplier;
+import java.util.stream.Collectors;
+import org.apache.hadoop.hdds.HddsConfigKeys;
+import org.apache.hadoop.hdds.client.BlockID;
+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.protocol.datanode.proto.ContainerProtos;
+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.ContainerTestHelper;
+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;
+import org.junit.jupiter.api.parallel.Execution;
+import org.junit.jupiter.api.parallel.ExecutionMode;
+import org.junit.jupiter.api.parallel.ResourceLock;
+
+/**
+ * Integration tests for per-volume push replication thread pools (HDDS-15412).
+ */
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
+@Execution(ExecutionMode.SAME_THREAD)
+@ResourceLock("MiniOzoneCluster")
+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 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 OzoneBucket bucket;
+
+ @BeforeAll
+ void setUp() throws Exception {
+ OzoneConfiguration conf = createSharedConfig();
+ cluster = newCluster(conf, DATANODE_COUNT);
+ cluster.waitForClusterToBeReady();
+ clientFactory = new XceiverClientManager(conf);
+ try (OzoneClient client = cluster.newClient()) {
+ bucket = DataTestUtil.createVolumeAndBucket(client, VOLUME, BUCKET);
+ }
+ }
+
+ @AfterAll
+ void tearDown() {
+ IOUtils.closeQuietly(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, 0L);
+
+ ReplicateContainerCommand cmd =
+ ReplicateContainerCommand.toTarget(containerId, target);
+ queuePushAndWaitForContainer(cluster, cmd, source, target, containerId);
+ getContainer(cluster, target, containerId);
+ assertVolumePools(sourceDn, DATA_VOLUMES, 1);
+
+ 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);
+
+ ContainerManager containerManager = cluster.getStorageContainerManager().getContainerManager();
+ Set<ContainerReplica> replicas =
+ containerManager.getContainerReplicas(ContainerID.valueOf(scmContainerId));
+ DatanodeDetails replicaDn = replicas.iterator().next().getDatanodeDetails();
+ cluster.shutdownHddsDatanode(replicaDn);
+ waitForReplicaCount(scmContainerId, 3, cluster);
+ }
+
+ @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, 0L);
+ long containerOnVol1 = findOrCreateContainerOnVolume(
+ cluster, clientFactory, source, vol1, 0L);
+
+ assertVolumePools(sourceDn, DATA_VOLUMES, 1);
+
+ triggerAndWaitForVolumeFailure(volSet, vol0);
+ waitForVolumePoolState(sourceDn, vol0, vol1);
+
+ ReplicateContainerCommand cmd =
+ ReplicateContainerCommand.toTarget(containerOnVol1, target);
+ queuePushAndWaitForContainer(cluster, cmd, source, target, containerOnVol1);
+ getContainer(cluster, target, containerOnVol1);
+ assertEquals(1, volSet.getFailedVolumesList().size());
+
+ // Task routes via global pool fallback (HDDS-15327); replication fails on bad volume.
+ ReplicateContainerCommand failedVolCmd =
+ ReplicateContainerCommand.toTarget(containerOnVol0, target);
+ ReplicationSupervisor supervisor =
+ sourceDn.getDatanodeStateMachine().getSupervisor();
+ long previousFailures = supervisor.getReplicationFailureCount();
+ queuePushAndWaitForFailure(cluster, failedVolCmd, source, target,
+ containerOnVol0, supervisor, previousFailures);
+ assertTrue(supervisor.getReplicationFailureCount() >= previousFailures + 1);
+
+ DatanodeTestUtils.restoreBadVolume(vol0);
+ }
+
+ @Order(3)
+ @Test
+ void testDecommissionWithPerVolumePools() throws Exception {
Review Comment:
Enables per-volume pools but asserts nothing per-volume. This mirrors `TestDecommissionAndMaintenance#testNodeWithOpenPipelineCanBeDecommissionedAndRecommissioned` and would pass with the feature off. What does it catch that the existing decommission test doesn't?
##########
hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/replication/TestPerVolumePushReplication.java:
##########
@@ -0,0 +1,545 @@
+/*
+ * 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.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.OzoneTestHelper.waitForReplicaCount;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.BooleanSupplier;
+import java.util.stream.Collectors;
+import org.apache.hadoop.hdds.HddsConfigKeys;
+import org.apache.hadoop.hdds.client.BlockID;
+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.protocol.datanode.proto.ContainerProtos;
+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.ContainerTestHelper;
+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;
+import org.junit.jupiter.api.parallel.Execution;
+import org.junit.jupiter.api.parallel.ExecutionMode;
+import org.junit.jupiter.api.parallel.ResourceLock;
+
+/**
+ * Integration tests for per-volume push replication thread pools (HDDS-15412).
+ */
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
+@Execution(ExecutionMode.SAME_THREAD)
+@ResourceLock("MiniOzoneCluster")
+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 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 OzoneBucket bucket;
+
+ @BeforeAll
+ void setUp() throws Exception {
+ OzoneConfiguration conf = createSharedConfig();
+ cluster = newCluster(conf, DATANODE_COUNT);
+ cluster.waitForClusterToBeReady();
+ clientFactory = new XceiverClientManager(conf);
+ try (OzoneClient client = cluster.newClient()) {
+ bucket = DataTestUtil.createVolumeAndBucket(client, VOLUME, BUCKET);
+ }
+ }
+
+ @AfterAll
+ void tearDown() {
+ IOUtils.closeQuietly(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, 0L);
+
+ ReplicateContainerCommand cmd =
+ ReplicateContainerCommand.toTarget(containerId, target);
+ queuePushAndWaitForContainer(cluster, cmd, source, target, containerId);
+ getContainer(cluster, target, containerId);
+ assertVolumePools(sourceDn, DATA_VOLUMES, 1);
+
+ 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);
+
+ ContainerManager containerManager = cluster.getStorageContainerManager().getContainerManager();
+ Set<ContainerReplica> replicas =
+ containerManager.getContainerReplicas(ContainerID.valueOf(scmContainerId));
+ DatanodeDetails replicaDn = replicas.iterator().next().getDatanodeDetails();
+ cluster.shutdownHddsDatanode(replicaDn);
+ waitForReplicaCount(scmContainerId, 3, cluster);
+ }
+
+ @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, 0L);
+ long containerOnVol1 = findOrCreateContainerOnVolume(
+ cluster, clientFactory, source, vol1, 0L);
+
+ assertVolumePools(sourceDn, DATA_VOLUMES, 1);
+
+ triggerAndWaitForVolumeFailure(volSet, vol0);
+ waitForVolumePoolState(sourceDn, vol0, vol1);
+
+ ReplicateContainerCommand cmd =
+ ReplicateContainerCommand.toTarget(containerOnVol1, target);
+ queuePushAndWaitForContainer(cluster, cmd, source, target, containerOnVol1);
+ getContainer(cluster, target, containerOnVol1);
+ assertEquals(1, volSet.getFailedVolumesList().size());
+
+ // Task routes via global pool fallback (HDDS-15327); replication fails on bad volume.
+ ReplicateContainerCommand failedVolCmd =
+ ReplicateContainerCommand.toTarget(containerOnVol0, target);
+ ReplicationSupervisor supervisor =
+ sourceDn.getDatanodeStateMachine().getSupervisor();
+ long previousFailures = supervisor.getReplicationFailureCount();
+ queuePushAndWaitForFailure(cluster, failedVolCmd, source, target,
+ containerOnVol0, supervisor, previousFailures);
+ assertTrue(supervisor.getReplicationFailureCount() >= previousFailures + 1);
+
+ DatanodeTestUtils.restoreBadVolume(vol0);
+ }
+
+ @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);
+
+ scmClient.decommissionNodes(singletonList(getDNHostAndPort(toDecommission)), false);
+ waitForDnToReachOpState(nm, toDecommission, DECOMMISSIONED);
+
+ 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(StandardCharsets.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) {
Review Comment:
This filters on op-state and failed-volume but not liveness, and `shutdownHddsDatanode` (L174, L247) leaves the node `IN_SERVICE` in `getHddsDatanodes()`. With the `@Order`ed shared cluster, could a later test pick a stopped node here and time out? Would a separate cluster per destructive test be safer?
##########
hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/replication/TestPerVolumePushReplication.java:
##########
@@ -0,0 +1,545 @@
+/*
+ * 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.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.OzoneTestHelper.waitForReplicaCount;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.BooleanSupplier;
+import java.util.stream.Collectors;
+import org.apache.hadoop.hdds.HddsConfigKeys;
+import org.apache.hadoop.hdds.client.BlockID;
+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.protocol.datanode.proto.ContainerProtos;
+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.ContainerTestHelper;
+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;
+import org.junit.jupiter.api.parallel.Execution;
+import org.junit.jupiter.api.parallel.ExecutionMode;
+import org.junit.jupiter.api.parallel.ResourceLock;
+
+/**
+ * Integration tests for per-volume push replication thread pools (HDDS-15412).
+ */
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
+@Execution(ExecutionMode.SAME_THREAD)
+@ResourceLock("MiniOzoneCluster")
+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 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 OzoneBucket bucket;
+
+ @BeforeAll
+ void setUp() throws Exception {
+ OzoneConfiguration conf = createSharedConfig();
+ cluster = newCluster(conf, DATANODE_COUNT);
+ cluster.waitForClusterToBeReady();
+ clientFactory = new XceiverClientManager(conf);
+ try (OzoneClient client = cluster.newClient()) {
+ bucket = DataTestUtil.createVolumeAndBucket(client, VOLUME, BUCKET);
+ }
+ }
+
+ @AfterAll
+ void tearDown() {
+ IOUtils.closeQuietly(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, 0L);
+
+ ReplicateContainerCommand cmd =
+ ReplicateContainerCommand.toTarget(containerId, target);
+ queuePushAndWaitForContainer(cluster, cmd, source, target, containerId);
+ getContainer(cluster, target, containerId);
+ assertVolumePools(sourceDn, DATA_VOLUMES, 1);
+
+ 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);
+
+ ContainerManager containerManager = cluster.getStorageContainerManager().getContainerManager();
+ Set<ContainerReplica> replicas =
+ containerManager.getContainerReplicas(ContainerID.valueOf(scmContainerId));
+ DatanodeDetails replicaDn = replicas.iterator().next().getDatanodeDetails();
+ cluster.shutdownHddsDatanode(replicaDn);
+ waitForReplicaCount(scmContainerId, 3, cluster);
+ }
+
+ @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, 0L);
+ long containerOnVol1 = findOrCreateContainerOnVolume(
+ cluster, clientFactory, source, vol1, 0L);
+
+ assertVolumePools(sourceDn, DATA_VOLUMES, 1);
+
+ triggerAndWaitForVolumeFailure(volSet, vol0);
+ waitForVolumePoolState(sourceDn, vol0, vol1);
+
+ ReplicateContainerCommand cmd =
+ ReplicateContainerCommand.toTarget(containerOnVol1, target);
+ queuePushAndWaitForContainer(cluster, cmd, source, target, containerOnVol1);
+ getContainer(cluster, target, containerOnVol1);
+ assertEquals(1, volSet.getFailedVolumesList().size());
+
+ // Task routes via global pool fallback (HDDS-15327); replication fails on bad volume.
+ ReplicateContainerCommand failedVolCmd =
+ ReplicateContainerCommand.toTarget(containerOnVol0, target);
+ ReplicationSupervisor supervisor =
+ sourceDn.getDatanodeStateMachine().getSupervisor();
+ long previousFailures = supervisor.getReplicationFailureCount();
+ queuePushAndWaitForFailure(cluster, failedVolCmd, source, target,
+ containerOnVol0, supervisor, previousFailures);
+ assertTrue(supervisor.getReplicationFailureCount() >= previousFailures + 1);
+
+ DatanodeTestUtils.restoreBadVolume(vol0);
+ }
+
+ @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);
+
+ scmClient.decommissionNodes(singletonList(getDNHostAndPort(toDecommission)), false);
+ waitForDnToReachOpState(nm, toDecommission, DECOMMISSIONED);
+
+ 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(StandardCharsets.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) {
+ 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 void assertVolumePools(HddsDatanodeService dn,
+ int expectedVolumeCount, int expectedPoolSize) {
+ VolumeReplicationThreadPools pools =
+ dn.getDatanodeStateMachine().getSupervisor().getVolumeReplicationThreadPools();
+ assertNotNull(pools);
+ List<? extends StorageVolume> volumes =
+ dn.getDatanodeStateMachine().getContainer().getVolumeSet().getVolumesList();
+ assertEquals(expectedVolumeCount, volumes.size());
+ for (StorageVolume volume : volumes) {
+ String volumeRoot = volume.getStorageDir().getPath();
+ assertTrue(pools.hasPool(volumeRoot), "missing pool for " + volumeRoot);
+ assertEquals(expectedPoolSize, pools.getPoolSize(volumeRoot));
+ }
+ }
+
+ 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((BooleanSupplier) () -> {
+ VolumeReplicationThreadPools pools =
+ sourceDn.getDatanodeStateMachine().getSupervisor().getVolumeReplicationThreadPools();
+ return pools != null
+ && !pools.hasPool(failedPath)
+ && pools.hasPool(healthyPath);
+ }, 100, 60000);
+ VolumeReplicationThreadPools pools =
+ sourceDn.getDatanodeStateMachine().getSupervisor().getVolumeReplicationThreadPools();
+ assertNotNull(pools);
+ assertFalse(pools.hasPool(failedPath));
+ assertTrue(pools.hasPool(healthyPath));
+ }
+
+ private static void queuePushAndWaitForContainer(MiniOzoneCluster cluster,
+ ReplicateContainerCommand cmd, DatanodeDetails source,
+ DatanodeDetails target, long containerId)
+ throws IOException, InterruptedException, TimeoutException {
+ queueReplicationCommand(cluster, cmd, source);
+ GenericTestUtils.waitFor((BooleanSupplier) () ->
+ hasContainer(cluster, target, containerId), 100, 30000);
+ }
+
+ private static void queuePushAndWaitForFailure(MiniOzoneCluster cluster,
+ ReplicateContainerCommand cmd, DatanodeDetails source,
+ DatanodeDetails target, long containerId, ReplicationSupervisor supervisor,
+ long previousFailureCount)
+ throws IOException, InterruptedException, TimeoutException {
+ queueReplicationCommand(cluster, cmd, source);
+ GenericTestUtils.waitFor((BooleanSupplier) () ->
+ supervisor.getReplicationFailureCount() >= previousFailureCount + 1
+ && !hasContainer(cluster, target, containerId),
+ 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);
Review Comment:
Identical to `TestContainerReplication#queueAndWaitForCompletion` (L267-268), and that same-package class also has `CONTAINER_ID` and `createNewClosedContainer`. Could we reuse those instead of re-deriving?
##########
hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/replication/TestPerVolumePushReplication.java:
##########
@@ -0,0 +1,545 @@
+/*
+ * 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.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.OzoneTestHelper.waitForReplicaCount;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.BooleanSupplier;
+import java.util.stream.Collectors;
+import org.apache.hadoop.hdds.HddsConfigKeys;
+import org.apache.hadoop.hdds.client.BlockID;
+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.protocol.datanode.proto.ContainerProtos;
+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.ContainerTestHelper;
+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;
+import org.junit.jupiter.api.parallel.Execution;
+import org.junit.jupiter.api.parallel.ExecutionMode;
+import org.junit.jupiter.api.parallel.ResourceLock;
+
+/**
+ * Integration tests for per-volume push replication thread pools (HDDS-15412).
+ */
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
+@Execution(ExecutionMode.SAME_THREAD)
+@ResourceLock("MiniOzoneCluster")
+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 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 OzoneBucket bucket;
+
+ @BeforeAll
+ void setUp() throws Exception {
+ OzoneConfiguration conf = createSharedConfig();
+ cluster = newCluster(conf, DATANODE_COUNT);
+ cluster.waitForClusterToBeReady();
+ clientFactory = new XceiverClientManager(conf);
+ try (OzoneClient client = cluster.newClient()) {
+ bucket = DataTestUtil.createVolumeAndBucket(client, VOLUME, BUCKET);
+ }
+ }
+
+ @AfterAll
+ void tearDown() {
+ IOUtils.closeQuietly(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, 0L);
+
+ ReplicateContainerCommand cmd =
+ ReplicateContainerCommand.toTarget(containerId, target);
+ queuePushAndWaitForContainer(cluster, cmd, source, target, containerId);
+ getContainer(cluster, target, containerId);
+ assertVolumePools(sourceDn, DATA_VOLUMES, 1);
+
+ 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);
+
+ ContainerManager containerManager = cluster.getStorageContainerManager().getContainerManager();
+ Set<ContainerReplica> replicas =
+ containerManager.getContainerReplicas(ContainerID.valueOf(scmContainerId));
+ DatanodeDetails replicaDn = replicas.iterator().next().getDatanodeDetails();
+ cluster.shutdownHddsDatanode(replicaDn);
+ waitForReplicaCount(scmContainerId, 3, cluster);
+ }
+
+ @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, 0L);
+ long containerOnVol1 = findOrCreateContainerOnVolume(
+ cluster, clientFactory, source, vol1, 0L);
+
+ assertVolumePools(sourceDn, DATA_VOLUMES, 1);
+
+ triggerAndWaitForVolumeFailure(volSet, vol0);
+ waitForVolumePoolState(sourceDn, vol0, vol1);
+
+ ReplicateContainerCommand cmd =
+ ReplicateContainerCommand.toTarget(containerOnVol1, target);
+ queuePushAndWaitForContainer(cluster, cmd, source, target, containerOnVol1);
+ getContainer(cluster, target, containerOnVol1);
+ assertEquals(1, volSet.getFailedVolumesList().size());
+
+ // Task routes via global pool fallback (HDDS-15327); replication fails on bad volume.
+ ReplicateContainerCommand failedVolCmd =
+ ReplicateContainerCommand.toTarget(containerOnVol0, target);
+ ReplicationSupervisor supervisor =
+ sourceDn.getDatanodeStateMachine().getSupervisor();
+ long previousFailures = supervisor.getReplicationFailureCount();
+ queuePushAndWaitForFailure(cluster, failedVolCmd, source, target,
+ containerOnVol0, supervisor, previousFailures);
+ assertTrue(supervisor.getReplicationFailureCount() >= previousFailures + 1);
+
+ DatanodeTestUtils.restoreBadVolume(vol0);
+ }
+
+ @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);
+
+ scmClient.decommissionNodes(singletonList(getDNHostAndPort(toDecommission)), false);
+ waitForDnToReachOpState(nm, toDecommission, DECOMMISSIONED);
+
+ 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(StandardCharsets.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) {
+ 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 void assertVolumePools(HddsDatanodeService dn,
+ int expectedVolumeCount, int expectedPoolSize) {
+ VolumeReplicationThreadPools pools =
+ dn.getDatanodeStateMachine().getSupervisor().getVolumeReplicationThreadPools();
+ assertNotNull(pools);
+ List<? extends StorageVolume> volumes =
+ dn.getDatanodeStateMachine().getContainer().getVolumeSet().getVolumesList();
+ assertEquals(expectedVolumeCount, volumes.size());
+ for (StorageVolume volume : volumes) {
+ String volumeRoot = volume.getStorageDir().getPath();
+ assertTrue(pools.hasPool(volumeRoot), "missing pool for " + volumeRoot);
+ assertEquals(expectedPoolSize, pools.getPoolSize(volumeRoot));
+ }
+ }
+
+ 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((BooleanSupplier) () -> {
+ VolumeReplicationThreadPools pools =
+ sourceDn.getDatanodeStateMachine().getSupervisor().getVolumeReplicationThreadPools();
+ return pools != null
+ && !pools.hasPool(failedPath)
+ && pools.hasPool(healthyPath);
+ }, 100, 60000);
+ VolumeReplicationThreadPools pools =
+ sourceDn.getDatanodeStateMachine().getSupervisor().getVolumeReplicationThreadPools();
+ assertNotNull(pools);
+ assertFalse(pools.hasPool(failedPath));
+ assertTrue(pools.hasPool(healthyPath));
+ }
+
+ private static void queuePushAndWaitForContainer(MiniOzoneCluster cluster,
+ ReplicateContainerCommand cmd, DatanodeDetails source,
+ DatanodeDetails target, long containerId)
+ throws IOException, InterruptedException, TimeoutException {
+ queueReplicationCommand(cluster, cmd, source);
+ GenericTestUtils.waitFor((BooleanSupplier) () ->
+ hasContainer(cluster, target, containerId), 100, 30000);
+ }
+
+ private static void queuePushAndWaitForFailure(MiniOzoneCluster cluster,
+ ReplicateContainerCommand cmd, DatanodeDetails source,
+ DatanodeDetails target, long containerId, ReplicationSupervisor supervisor,
+ long previousFailureCount)
+ throws IOException, InterruptedException, TimeoutException {
+ queueReplicationCommand(cluster, cmd, source);
+ GenericTestUtils.waitFor((BooleanSupplier) () ->
+ supervisor.getReplicationFailureCount() >= previousFailureCount + 1
+ && !hasContainer(cluster, target, containerId),
+ 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,
+ long dataSize) throws Exception {
+ for (int attempt = 0; attempt < 30; attempt++) {
+ long containerId = createClosedContainer(clientFactory, dn, dataSize);
+ 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, long dataSize) throws Exception {
+ long containerId = CONTAINER_ID.incrementAndGet();
+ try (XceiverClientSpi client = clientFactory.acquireClient(createPipeline(singleton(dn)))) {
+ if (dataSize <= 0) {
+ createContainer(client, containerId, null, CLOSED, 0);
+ return containerId;
+ }
+ createContainer(client, containerId, null);
Review Comment:
This branch looks unreachable, since all callers (L157, L188, L190) pass `dataSize=0`. Removing it makes this identical to `TestContainerReplication#createNewClosedContainer`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-15802. Add integration tests for per-volume push replication [ozone]
Posted by "smengcl (via GitHub)" <gi...@apache.org>.
smengcl commented on code in PR #11146:
URL: https://github.com/apache/ozone/pull/11146#discussion_r4100708258
##########
hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/replication/TestPerVolumePushReplication.java:
##########
@@ -0,0 +1,545 @@
+/*
+ * 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.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.OzoneTestHelper.waitForReplicaCount;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.BooleanSupplier;
+import java.util.stream.Collectors;
+import org.apache.hadoop.hdds.HddsConfigKeys;
+import org.apache.hadoop.hdds.client.BlockID;
+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.protocol.datanode.proto.ContainerProtos;
+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.ContainerTestHelper;
+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;
+import org.junit.jupiter.api.parallel.Execution;
+import org.junit.jupiter.api.parallel.ExecutionMode;
+import org.junit.jupiter.api.parallel.ResourceLock;
+
+/**
+ * Integration tests for per-volume push replication thread pools (HDDS-15412).
+ */
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
+@Execution(ExecutionMode.SAME_THREAD)
+@ResourceLock("MiniOzoneCluster")
+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 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 OzoneBucket bucket;
+
+ @BeforeAll
+ void setUp() throws Exception {
+ OzoneConfiguration conf = createSharedConfig();
+ cluster = newCluster(conf, DATANODE_COUNT);
+ cluster.waitForClusterToBeReady();
+ clientFactory = new XceiverClientManager(conf);
+ try (OzoneClient client = cluster.newClient()) {
+ bucket = DataTestUtil.createVolumeAndBucket(client, VOLUME, BUCKET);
+ }
+ }
+
+ @AfterAll
+ void tearDown() {
+ IOUtils.closeQuietly(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, 0L);
+
+ ReplicateContainerCommand cmd =
+ ReplicateContainerCommand.toTarget(containerId, target);
+ queuePushAndWaitForContainer(cluster, cmd, source, target, containerId);
+ getContainer(cluster, target, containerId);
+ assertVolumePools(sourceDn, DATA_VOLUMES, 1);
+
+ 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);
+
+ ContainerManager containerManager = cluster.getStorageContainerManager().getContainerManager();
+ Set<ContainerReplica> replicas =
+ containerManager.getContainerReplicas(ContainerID.valueOf(scmContainerId));
+ DatanodeDetails replicaDn = replicas.iterator().next().getDatanodeDetails();
+ cluster.shutdownHddsDatanode(replicaDn);
+ waitForReplicaCount(scmContainerId, 3, cluster);
+ }
+
+ @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, 0L);
+ long containerOnVol1 = findOrCreateContainerOnVolume(
+ cluster, clientFactory, source, vol1, 0L);
+
+ assertVolumePools(sourceDn, DATA_VOLUMES, 1);
+
+ triggerAndWaitForVolumeFailure(volSet, vol0);
+ waitForVolumePoolState(sourceDn, vol0, vol1);
+
+ ReplicateContainerCommand cmd =
+ ReplicateContainerCommand.toTarget(containerOnVol1, target);
+ queuePushAndWaitForContainer(cluster, cmd, source, target, containerOnVol1);
+ getContainer(cluster, target, containerOnVol1);
+ assertEquals(1, volSet.getFailedVolumesList().size());
+
+ // Task routes via global pool fallback (HDDS-15327); replication fails on bad volume.
+ ReplicateContainerCommand failedVolCmd =
+ ReplicateContainerCommand.toTarget(containerOnVol0, target);
+ ReplicationSupervisor supervisor =
+ sourceDn.getDatanodeStateMachine().getSupervisor();
+ long previousFailures = supervisor.getReplicationFailureCount();
+ queuePushAndWaitForFailure(cluster, failedVolCmd, source, target,
+ containerOnVol0, supervisor, previousFailures);
+ assertTrue(supervisor.getReplicationFailureCount() >= previousFailures + 1);
+
+ DatanodeTestUtils.restoreBadVolume(vol0);
+ }
+
+ @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);
+
+ scmClient.decommissionNodes(singletonList(getDNHostAndPort(toDecommission)), false);
+ waitForDnToReachOpState(nm, toDecommission, DECOMMISSIONED);
+
+ 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(StandardCharsets.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) {
Review Comment:
Yes, it could, and it was reachable today. Addressed in 66bf80e1221.
`selectHealthyDatanode` now rejects stopped nodes via `HddsDatanodeService#isStopped`, so a leaked shutdown fails on selection with a readable message instead of timing out inside a later wait. On top of that the first test restarts the datanode it stops, and `restoreBadVolume` moved into a `finally` block, so neither destructive test can leave state behind even when an assertion fails partway through. The class javadoc now states that contract.
On a separate cluster per destructive test: I kept the shared cluster. Three `MiniOzoneCluster` lifecycles at 7 datanodes each is a large runtime cost for this class, and the leaked state was the actual defect rather than the sharing. If you would still rather have the isolation, say so and I will switch to `MiniOzoneClusterProvider`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-15802. Add integration tests for per-volume push replication [ozone]
Posted by "smengcl (via GitHub)" <gi...@apache.org>.
smengcl commented on code in PR #11146:
URL: https://github.com/apache/ozone/pull/11146#discussion_r4100705536
##########
hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/replication/TestPerVolumePushReplication.java:
##########
@@ -0,0 +1,526 @@
+/*
+ * 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.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.OzoneTestHelper.waitForReplicaCount;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.BooleanSupplier;
+import java.util.function.ToLongFunction;
+import java.util.stream.Collectors;
+import org.apache.hadoop.hdds.HddsConfigKeys;
+import org.apache.hadoop.hdds.client.BlockID;
+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.protocol.datanode.proto.ContainerProtos;
+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.HddsDatanodeService;
+import org.apache.hadoop.ozone.MiniOzoneCluster;
+import org.apache.hadoop.ozone.OzoneConfigKeys;
+import org.apache.hadoop.ozone.DataTestUtil;
+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.ContainerTestHelper;
+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;
+import org.junit.jupiter.api.parallel.Execution;
+import org.junit.jupiter.api.parallel.ExecutionMode;
+import org.junit.jupiter.api.parallel.ResourceLock;
+
+/**
+ * Integration tests for per-volume push replication thread pools (HDDS-15412).
+ */
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
+@Execution(ExecutionMode.SAME_THREAD)
+@ResourceLock("MiniOzoneCluster")
+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 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;
+
+ @BeforeAll
+ void setUp() throws Exception {
+ OzoneConfiguration conf = createSharedConfig();
+ cluster = newCluster(conf, DATANODE_COUNT);
+ cluster.waitForClusterToBeReady();
+ clientFactory = new XceiverClientManager(conf);
+ }
+
+ @AfterAll
+ void tearDown() {
+ IOUtils.closeQuietly(clientFactory, cluster);
+ }
+
+ @Order(1)
+ @Test
+ void testPushAndScmReplicationWithPerVolumeEnabled() throws Exception {
+ try (OzoneClient client = cluster.newClient()) {
+ HddsDatanodeService sourceDn = selectHealthyDatanode(0);
+ DatanodeDetails source = sourceDn.getDatanodeDetails();
+ DatanodeDetails target = selectOtherHealthyNode(source);
+ long containerId = createClosedContainer(clientFactory, source, 0L);
+
+ ReplicateContainerCommand cmd =
+ ReplicateContainerCommand.toTarget(containerId, target);
+ queuePushAndWait(cluster, cmd, source, ReplicationSupervisor::getReplicationSuccessCount);
+ assertNotNull(getContainer(cluster, target, containerId));
+ assertVolumePools(sourceDn, DATA_VOLUMES, 1);
+
+ OzoneBucket bucket = DataTestUtil.createVolumeAndBucket(client, VOLUME, BUCKET);
+ 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);
+
+ ContainerManager containerManager = cluster.getStorageContainerManager().getContainerManager();
+ Set<ContainerReplica> replicas =
+ containerManager.getContainerReplicas(ContainerID.valueOf(scmContainerId));
+ DatanodeDetails replicaDn = replicas.iterator().next().getDatanodeDetails();
+ cluster.shutdownHddsDatanode(replicaDn);
+ waitForReplicaCount(scmContainerId, 3, cluster);
+ }
+ }
+
+ @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, 0L);
+ long containerOnVol1 = findOrCreateContainerOnVolume(
+ cluster, clientFactory, source, vol1, 0L);
+
+ assertVolumePools(sourceDn, DATA_VOLUMES, 1);
+
+ triggerAndWaitForVolumeFailure(volSet, vol0);
+ waitForVolumePoolState(sourceDn, vol0, vol1);
+
+ ReplicateContainerCommand cmd =
+ ReplicateContainerCommand.toTarget(containerOnVol1, target);
+ queuePushAndWait(cluster, cmd, source, ReplicationSupervisor::getReplicationSuccessCount);
+ assertNotNull(getContainer(cluster, target, containerOnVol1));
+ assertEquals(1, volSet.getFailedVolumesList().size());
+
+ // Task routes via global pool fallback (HDDS-15327); replication fails on bad volume.
+ ReplicateContainerCommand failedVolCmd =
+ ReplicateContainerCommand.toTarget(containerOnVol0, target);
+ ReplicationSupervisor supervisor =
+ sourceDn.getDatanodeStateMachine().getSupervisor();
+ long previousFailures = supervisor.getReplicationFailureCount();
+ queuePushAndWait(cluster, failedVolCmd, source,
+ ReplicationSupervisor::getReplicationFailureCount);
+ assertTrue(supervisor.getReplicationFailureCount() >= previousFailures + 1);
+
+ DatanodeTestUtils.restoreBadVolume(vol0);
+ }
+
+ @Order(3)
+ @Test
+ void testDecommissionWithPerVolumePools() throws Exception {
+ try (OzoneClient client = cluster.newClient();
+ ContainerOperationClient scmClient = new ContainerOperationClient(cluster.getConf())) {
+ StorageContainerManager scm = cluster.getStorageContainerManager();
+ NodeManager nm = scm.getScmNodeManager();
+ ContainerManager cm = scm.getContainerManager();
+ PipelineManager pm = scm.getPipelineManager();
+
+ OzoneBucket bucket = client.getObjectStore().getVolume(VOLUME).getBucket(BUCKET);
+ 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);
+
+ scmClient.decommissionNodes(singletonList(getDNHostAndPort(toDecommission)), false);
+ waitForDnToReachOpState(nm, toDecommission, DECOMMISSIONED);
+
+ 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(StandardCharsets.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) {
+ 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 void assertVolumePools(HddsDatanodeService dn,
+ int expectedVolumeCount, int expectedPoolSize) {
+ VolumeReplicationThreadPools pools =
+ dn.getDatanodeStateMachine().getSupervisor().getVolumeReplicationThreadPools();
+ assertNotNull(pools);
+ List<? extends StorageVolume> volumes =
+ dn.getDatanodeStateMachine().getContainer().getVolumeSet().getVolumesList();
+ assertEquals(expectedVolumeCount, volumes.size());
+ for (StorageVolume volume : volumes) {
+ String volumeRoot = volume.getStorageDir().getPath();
+ assertTrue(pools.hasPool(volumeRoot), "missing pool for " + volumeRoot);
+ assertEquals(expectedPoolSize, pools.getPoolSize(volumeRoot));
+ }
+ }
+
+ 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((BooleanSupplier) () -> {
+ VolumeReplicationThreadPools pools =
+ sourceDn.getDatanodeStateMachine().getSupervisor().getVolumeReplicationThreadPools();
+ return pools != null
+ && !pools.hasPool(failedPath)
+ && pools.hasPool(healthyPath);
+ }, 100, 60000);
+ VolumeReplicationThreadPools pools =
+ sourceDn.getDatanodeStateMachine().getSupervisor().getVolumeReplicationThreadPools();
+ assertNotNull(pools);
+ assertFalse(pools.hasPool(failedPath));
+ assertTrue(pools.hasPool(healthyPath));
+ }
+
+ private static void queuePushAndWait(MiniOzoneCluster cluster,
+ ReplicateContainerCommand cmd, DatanodeDetails source,
+ ToLongFunction<ReplicationSupervisor> counter)
+ throws IOException, InterruptedException, TimeoutException {
+ DatanodeStateMachine stateMachine = cluster.getHddsDatanode(source).getDatanodeStateMachine();
+ ReplicationSupervisor supervisor = stateMachine.getSupervisor();
+ long previousCount = counter.applyAsLong(supervisor);
+ long targetCount = previousCount + 1;
+ StateContext context = stateMachine.getContext();
+ context.getTermOfLeaderSCM().ifPresent(cmd::setTerm);
+ context.addCommand(cmd);
+ GenericTestUtils.waitFor((BooleanSupplier) () ->
+ counter.applyAsLong(supervisor) >= targetCount, 100, 30000);
+ }
+
+ private static long findOrCreateContainerOnVolume(MiniOzoneCluster cluster,
+ XceiverClientFactory clientFactory, DatanodeDetails dn, HddsVolume targetVolume,
+ long dataSize) throws Exception {
+ for (int attempt = 0; attempt < 30; attempt++) {
+ long containerId = createClosedContainer(clientFactory, dn, dataSize);
+ 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, long dataSize) throws Exception {
+ long containerId = CONTAINER_ID.incrementAndGet();
+ try (XceiverClientSpi client = clientFactory.acquireClient(createPipeline(singleton(dn)))) {
+ if (dataSize <= 0) {
+ createContainer(client, containerId, null, CLOSED, 0);
+ return containerId;
+ }
+ createContainer(client, containerId, null);
+ int chunkSize = 1024 * 1024;
+ long totalBytesWritten = 0;
+ while (totalBytesWritten < dataSize) {
+ BlockID blockId = ContainerTestHelper.getTestBlockID(containerId);
+ long remainingBytes = dataSize - totalBytesWritten;
+ int currentChunkSize = (int) Math.min(chunkSize, remainingBytes);
+ ContainerProtos.ContainerCommandRequestProto writeChunkRequest =
+ ContainerTestHelper.getWriteChunkRequest(
+ createPipeline(singleton(dn)), blockId, currentChunkSize);
+ client.sendCommand(writeChunkRequest);
+ ContainerProtos.ContainerCommandRequestProto putBlockRequest =
+ ContainerTestHelper.getPutBlockRequest(writeChunkRequest);
+ client.sendCommand(putBlockRequest);
+ totalBytesWritten += currentChunkSize;
+ }
+ ContainerProtos.CloseContainerRequestProto closeRequest =
+ ContainerProtos.CloseContainerRequestProto.newBuilder().build();
+ ContainerProtos.ContainerCommandRequestProto closeContainerRequest =
+ ContainerProtos.ContainerCommandRequestProto.newBuilder()
+ .setCmdType(ContainerProtos.Type.CloseContainer)
+ .setContainerID(containerId)
+ .setCloseContainer(closeRequest)
+ .setDatanodeUuid(dn.getUuidString())
+ .build();
+ client.sendCommand(closeContainerRequest);
+ 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;
+ }
Review Comment:
Addressed in 66bf80e1221.
The `assertNotNull` wrappers were dropped in an earlier commit on this branch, which left bare `getContainer(...)` calls whose result went unused right after a `queuePushAndWaitForContainer(...)` that had already confirmed presence via `hasContainer`. That commit removes those leftover calls. The two `getContainer(...)` calls that remain both use the return value.
##########
hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/replication/TestPerVolumePushReplication.java:
##########
@@ -0,0 +1,526 @@
+/*
+ * 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.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.OzoneTestHelper.waitForReplicaCount;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.BooleanSupplier;
+import java.util.function.ToLongFunction;
+import java.util.stream.Collectors;
+import org.apache.hadoop.hdds.HddsConfigKeys;
+import org.apache.hadoop.hdds.client.BlockID;
+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.protocol.datanode.proto.ContainerProtos;
+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.HddsDatanodeService;
+import org.apache.hadoop.ozone.MiniOzoneCluster;
+import org.apache.hadoop.ozone.OzoneConfigKeys;
+import org.apache.hadoop.ozone.DataTestUtil;
+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.ContainerTestHelper;
+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;
+import org.junit.jupiter.api.parallel.Execution;
+import org.junit.jupiter.api.parallel.ExecutionMode;
+import org.junit.jupiter.api.parallel.ResourceLock;
+
+/**
+ * Integration tests for per-volume push replication thread pools (HDDS-15412).
+ */
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
+@Execution(ExecutionMode.SAME_THREAD)
+@ResourceLock("MiniOzoneCluster")
+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 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;
+
+ @BeforeAll
+ void setUp() throws Exception {
+ OzoneConfiguration conf = createSharedConfig();
+ cluster = newCluster(conf, DATANODE_COUNT);
+ cluster.waitForClusterToBeReady();
+ clientFactory = new XceiverClientManager(conf);
+ }
+
+ @AfterAll
+ void tearDown() {
+ IOUtils.closeQuietly(clientFactory, cluster);
+ }
+
+ @Order(1)
+ @Test
+ void testPushAndScmReplicationWithPerVolumeEnabled() throws Exception {
+ try (OzoneClient client = cluster.newClient()) {
+ HddsDatanodeService sourceDn = selectHealthyDatanode(0);
+ DatanodeDetails source = sourceDn.getDatanodeDetails();
+ DatanodeDetails target = selectOtherHealthyNode(source);
+ long containerId = createClosedContainer(clientFactory, source, 0L);
+
+ ReplicateContainerCommand cmd =
+ ReplicateContainerCommand.toTarget(containerId, target);
+ queuePushAndWait(cluster, cmd, source, ReplicationSupervisor::getReplicationSuccessCount);
+ assertNotNull(getContainer(cluster, target, containerId));
+ assertVolumePools(sourceDn, DATA_VOLUMES, 1);
+
+ OzoneBucket bucket = DataTestUtil.createVolumeAndBucket(client, VOLUME, BUCKET);
+ 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);
+
+ ContainerManager containerManager = cluster.getStorageContainerManager().getContainerManager();
+ Set<ContainerReplica> replicas =
+ containerManager.getContainerReplicas(ContainerID.valueOf(scmContainerId));
+ DatanodeDetails replicaDn = replicas.iterator().next().getDatanodeDetails();
+ cluster.shutdownHddsDatanode(replicaDn);
+ waitForReplicaCount(scmContainerId, 3, cluster);
+ }
+ }
+
+ @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, 0L);
+ long containerOnVol1 = findOrCreateContainerOnVolume(
+ cluster, clientFactory, source, vol1, 0L);
+
+ assertVolumePools(sourceDn, DATA_VOLUMES, 1);
+
+ triggerAndWaitForVolumeFailure(volSet, vol0);
+ waitForVolumePoolState(sourceDn, vol0, vol1);
+
+ ReplicateContainerCommand cmd =
+ ReplicateContainerCommand.toTarget(containerOnVol1, target);
+ queuePushAndWait(cluster, cmd, source, ReplicationSupervisor::getReplicationSuccessCount);
+ assertNotNull(getContainer(cluster, target, containerOnVol1));
+ assertEquals(1, volSet.getFailedVolumesList().size());
+
+ // Task routes via global pool fallback (HDDS-15327); replication fails on bad volume.
+ ReplicateContainerCommand failedVolCmd =
+ ReplicateContainerCommand.toTarget(containerOnVol0, target);
+ ReplicationSupervisor supervisor =
+ sourceDn.getDatanodeStateMachine().getSupervisor();
+ long previousFailures = supervisor.getReplicationFailureCount();
+ queuePushAndWait(cluster, failedVolCmd, source,
+ ReplicationSupervisor::getReplicationFailureCount);
+ assertTrue(supervisor.getReplicationFailureCount() >= previousFailures + 1);
+
+ DatanodeTestUtils.restoreBadVolume(vol0);
+ }
+
+ @Order(3)
+ @Test
+ void testDecommissionWithPerVolumePools() throws Exception {
+ try (OzoneClient client = cluster.newClient();
+ ContainerOperationClient scmClient = new ContainerOperationClient(cluster.getConf())) {
+ StorageContainerManager scm = cluster.getStorageContainerManager();
+ NodeManager nm = scm.getScmNodeManager();
+ ContainerManager cm = scm.getContainerManager();
+ PipelineManager pm = scm.getPipelineManager();
+
+ OzoneBucket bucket = client.getObjectStore().getVolume(VOLUME).getBucket(BUCKET);
+ 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);
+
+ scmClient.decommissionNodes(singletonList(getDNHostAndPort(toDecommission)), false);
+ waitForDnToReachOpState(nm, toDecommission, DECOMMISSIONED);
+
+ 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(StandardCharsets.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) {
+ 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 void assertVolumePools(HddsDatanodeService dn,
+ int expectedVolumeCount, int expectedPoolSize) {
+ VolumeReplicationThreadPools pools =
+ dn.getDatanodeStateMachine().getSupervisor().getVolumeReplicationThreadPools();
+ assertNotNull(pools);
+ List<? extends StorageVolume> volumes =
+ dn.getDatanodeStateMachine().getContainer().getVolumeSet().getVolumesList();
+ assertEquals(expectedVolumeCount, volumes.size());
+ for (StorageVolume volume : volumes) {
+ String volumeRoot = volume.getStorageDir().getPath();
+ assertTrue(pools.hasPool(volumeRoot), "missing pool for " + volumeRoot);
+ assertEquals(expectedPoolSize, pools.getPoolSize(volumeRoot));
+ }
+ }
+
+ 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((BooleanSupplier) () -> {
+ VolumeReplicationThreadPools pools =
+ sourceDn.getDatanodeStateMachine().getSupervisor().getVolumeReplicationThreadPools();
+ return pools != null
+ && !pools.hasPool(failedPath)
+ && pools.hasPool(healthyPath);
+ }, 100, 60000);
+ VolumeReplicationThreadPools pools =
+ sourceDn.getDatanodeStateMachine().getSupervisor().getVolumeReplicationThreadPools();
+ assertNotNull(pools);
+ assertFalse(pools.hasPool(failedPath));
+ assertTrue(pools.hasPool(healthyPath));
+ }
+
+ private static void queuePushAndWait(MiniOzoneCluster cluster,
+ ReplicateContainerCommand cmd, DatanodeDetails source,
+ ToLongFunction<ReplicationSupervisor> counter)
+ throws IOException, InterruptedException, TimeoutException {
+ DatanodeStateMachine stateMachine = cluster.getHddsDatanode(source).getDatanodeStateMachine();
+ ReplicationSupervisor supervisor = stateMachine.getSupervisor();
+ long previousCount = counter.applyAsLong(supervisor);
+ long targetCount = previousCount + 1;
+ StateContext context = stateMachine.getContext();
+ context.getTermOfLeaderSCM().ifPresent(cmd::setTerm);
+ context.addCommand(cmd);
+ GenericTestUtils.waitFor((BooleanSupplier) () ->
+ counter.applyAsLong(supervisor) >= targetCount, 100, 30000);
Review Comment:
Addressed in 66bf80e1221. This thread reads as outdated but the issue was still present, so it is worth answering.
The failure wait and the assertion now use `supervisor.getReplicationFailureCount(ReplicationTask.METRIC_NAME)` instead of the cluster wide counter, so unrelated SCM driven work on the same datanode can no longer satisfy the condition on its own. The expected failure case additionally asserts that `hasContainer(target, containerOnVol0)` is false, which is the container specific check you suggested, and the success path already polls until the target datanode actually holds the container.
##########
hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/replication/TestPerVolumePushReplication.java:
##########
@@ -0,0 +1,545 @@
+/*
+ * 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.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.OzoneTestHelper.waitForReplicaCount;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.BooleanSupplier;
+import java.util.stream.Collectors;
+import org.apache.hadoop.hdds.HddsConfigKeys;
+import org.apache.hadoop.hdds.client.BlockID;
+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.protocol.datanode.proto.ContainerProtos;
+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.ContainerTestHelper;
+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;
+import org.junit.jupiter.api.parallel.Execution;
+import org.junit.jupiter.api.parallel.ExecutionMode;
+import org.junit.jupiter.api.parallel.ResourceLock;
+
+/**
+ * Integration tests for per-volume push replication thread pools (HDDS-15412).
+ */
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
+@Execution(ExecutionMode.SAME_THREAD)
+@ResourceLock("MiniOzoneCluster")
+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 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 OzoneBucket bucket;
+
+ @BeforeAll
+ void setUp() throws Exception {
+ OzoneConfiguration conf = createSharedConfig();
+ cluster = newCluster(conf, DATANODE_COUNT);
+ cluster.waitForClusterToBeReady();
+ clientFactory = new XceiverClientManager(conf);
+ try (OzoneClient client = cluster.newClient()) {
+ bucket = DataTestUtil.createVolumeAndBucket(client, VOLUME, BUCKET);
+ }
+ }
+
+ @AfterAll
+ void tearDown() {
+ IOUtils.closeQuietly(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, 0L);
+
+ ReplicateContainerCommand cmd =
+ ReplicateContainerCommand.toTarget(containerId, target);
+ queuePushAndWaitForContainer(cluster, cmd, source, target, containerId);
+ getContainer(cluster, target, containerId);
+ assertVolumePools(sourceDn, DATA_VOLUMES, 1);
+
+ 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);
+
+ ContainerManager containerManager = cluster.getStorageContainerManager().getContainerManager();
+ Set<ContainerReplica> replicas =
+ containerManager.getContainerReplicas(ContainerID.valueOf(scmContainerId));
+ DatanodeDetails replicaDn = replicas.iterator().next().getDatanodeDetails();
+ cluster.shutdownHddsDatanode(replicaDn);
+ waitForReplicaCount(scmContainerId, 3, cluster);
+ }
+
+ @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, 0L);
+ long containerOnVol1 = findOrCreateContainerOnVolume(
+ cluster, clientFactory, source, vol1, 0L);
+
+ assertVolumePools(sourceDn, DATA_VOLUMES, 1);
+
+ triggerAndWaitForVolumeFailure(volSet, vol0);
+ waitForVolumePoolState(sourceDn, vol0, vol1);
+
+ ReplicateContainerCommand cmd =
+ ReplicateContainerCommand.toTarget(containerOnVol1, target);
+ queuePushAndWaitForContainer(cluster, cmd, source, target, containerOnVol1);
+ getContainer(cluster, target, containerOnVol1);
+ assertEquals(1, volSet.getFailedVolumesList().size());
+
+ // Task routes via global pool fallback (HDDS-15327); replication fails on bad volume.
+ ReplicateContainerCommand failedVolCmd =
+ ReplicateContainerCommand.toTarget(containerOnVol0, target);
+ ReplicationSupervisor supervisor =
+ sourceDn.getDatanodeStateMachine().getSupervisor();
+ long previousFailures = supervisor.getReplicationFailureCount();
+ queuePushAndWaitForFailure(cluster, failedVolCmd, source, target,
+ containerOnVol0, supervisor, previousFailures);
+ assertTrue(supervisor.getReplicationFailureCount() >= previousFailures + 1);
+
+ DatanodeTestUtils.restoreBadVolume(vol0);
+ }
+
+ @Order(3)
+ @Test
+ void testDecommissionWithPerVolumePools() throws Exception {
Review Comment:
You were right, the test asserted nothing per volume. Addressed in 66bf80e1221.
The decommission test now asserts the one path `TestDecommissionAndMaintenance` cannot reach. Once the node reaches `DECOMMISSIONED`, SCM sends `SetNodeOperationalState`, the handler calls `supervisor.nodeStateUpdated`, and the per volume pools resize to `scaleOutOfServiceLimit(PER_VOLUME_STREAMS)`. The test waits for every volume pool on that datanode to reach that size, so it fails with the feature off (no per volume pools exist to resize) and it fails if the resize path breaks.
The other two tests had the same gap. They now capture `ThreadPoolExecutor.getCompletedTaskCount()` on the pool of the volume that holds the container, push, and wait for that specific counter to advance. So a push served entirely by the global pool no longer passes. `VolumeReplicationThreadPools#getExecutor` is package private and this test shares the package, so this needed no production change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-15802. Add integration tests for per-volume push replication [ozone]
Posted by "smengcl (via GitHub)" <gi...@apache.org>.
smengcl merged PR #11146:
URL: https://github.com/apache/ozone/pull/11146
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-15802. Add integration tests for per-volume push replication [ozone]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #11146:
URL: https://github.com/apache/ozone/pull/11146#discussion_r3882968480
##########
hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/replication/TestPerVolumePushReplication.java:
##########
@@ -0,0 +1,526 @@
+/*
+ * 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.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.OzoneTestHelper.waitForReplicaCount;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.BooleanSupplier;
+import java.util.function.ToLongFunction;
+import java.util.stream.Collectors;
+import org.apache.hadoop.hdds.HddsConfigKeys;
+import org.apache.hadoop.hdds.client.BlockID;
+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.protocol.datanode.proto.ContainerProtos;
+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.HddsDatanodeService;
+import org.apache.hadoop.ozone.MiniOzoneCluster;
+import org.apache.hadoop.ozone.OzoneConfigKeys;
+import org.apache.hadoop.ozone.DataTestUtil;
+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.ContainerTestHelper;
+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;
+import org.junit.jupiter.api.parallel.Execution;
+import org.junit.jupiter.api.parallel.ExecutionMode;
+import org.junit.jupiter.api.parallel.ResourceLock;
+
+/**
+ * Integration tests for per-volume push replication thread pools (HDDS-15412).
+ */
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
+@Execution(ExecutionMode.SAME_THREAD)
+@ResourceLock("MiniOzoneCluster")
+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 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;
+
+ @BeforeAll
+ void setUp() throws Exception {
+ OzoneConfiguration conf = createSharedConfig();
+ cluster = newCluster(conf, DATANODE_COUNT);
+ cluster.waitForClusterToBeReady();
+ clientFactory = new XceiverClientManager(conf);
+ }
+
+ @AfterAll
+ void tearDown() {
+ IOUtils.closeQuietly(clientFactory, cluster);
+ }
+
+ @Order(1)
+ @Test
+ void testPushAndScmReplicationWithPerVolumeEnabled() throws Exception {
+ try (OzoneClient client = cluster.newClient()) {
+ HddsDatanodeService sourceDn = selectHealthyDatanode(0);
+ DatanodeDetails source = sourceDn.getDatanodeDetails();
+ DatanodeDetails target = selectOtherHealthyNode(source);
+ long containerId = createClosedContainer(clientFactory, source, 0L);
+
+ ReplicateContainerCommand cmd =
+ ReplicateContainerCommand.toTarget(containerId, target);
+ queuePushAndWait(cluster, cmd, source, ReplicationSupervisor::getReplicationSuccessCount);
+ assertNotNull(getContainer(cluster, target, containerId));
+ assertVolumePools(sourceDn, DATA_VOLUMES, 1);
+
+ OzoneBucket bucket = DataTestUtil.createVolumeAndBucket(client, VOLUME, BUCKET);
+ 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);
+
+ ContainerManager containerManager = cluster.getStorageContainerManager().getContainerManager();
+ Set<ContainerReplica> replicas =
+ containerManager.getContainerReplicas(ContainerID.valueOf(scmContainerId));
+ DatanodeDetails replicaDn = replicas.iterator().next().getDatanodeDetails();
+ cluster.shutdownHddsDatanode(replicaDn);
+ waitForReplicaCount(scmContainerId, 3, cluster);
+ }
+ }
+
+ @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, 0L);
+ long containerOnVol1 = findOrCreateContainerOnVolume(
+ cluster, clientFactory, source, vol1, 0L);
+
+ assertVolumePools(sourceDn, DATA_VOLUMES, 1);
+
+ triggerAndWaitForVolumeFailure(volSet, vol0);
+ waitForVolumePoolState(sourceDn, vol0, vol1);
+
+ ReplicateContainerCommand cmd =
+ ReplicateContainerCommand.toTarget(containerOnVol1, target);
+ queuePushAndWait(cluster, cmd, source, ReplicationSupervisor::getReplicationSuccessCount);
+ assertNotNull(getContainer(cluster, target, containerOnVol1));
+ assertEquals(1, volSet.getFailedVolumesList().size());
+
+ // Task routes via global pool fallback (HDDS-15327); replication fails on bad volume.
+ ReplicateContainerCommand failedVolCmd =
+ ReplicateContainerCommand.toTarget(containerOnVol0, target);
+ ReplicationSupervisor supervisor =
+ sourceDn.getDatanodeStateMachine().getSupervisor();
+ long previousFailures = supervisor.getReplicationFailureCount();
+ queuePushAndWait(cluster, failedVolCmd, source,
+ ReplicationSupervisor::getReplicationFailureCount);
+ assertTrue(supervisor.getReplicationFailureCount() >= previousFailures + 1);
+
+ DatanodeTestUtils.restoreBadVolume(vol0);
+ }
+
+ @Order(3)
+ @Test
+ void testDecommissionWithPerVolumePools() throws Exception {
+ try (OzoneClient client = cluster.newClient();
+ ContainerOperationClient scmClient = new ContainerOperationClient(cluster.getConf())) {
+ StorageContainerManager scm = cluster.getStorageContainerManager();
+ NodeManager nm = scm.getScmNodeManager();
+ ContainerManager cm = scm.getContainerManager();
+ PipelineManager pm = scm.getPipelineManager();
+
+ OzoneBucket bucket = client.getObjectStore().getVolume(VOLUME).getBucket(BUCKET);
Review Comment:
This test assumes `VOLUME`/`BUCKET` already exist (created in `testPushAndScmReplicationWithPerVolumeEnabled`), creating an order-dependent test that can fail when run in isolation (eg `-Dtest=...#testDecommissionWithPerVolumePools`). Make the test self-contained by creating the volume/bucket in `@BeforeAll` (shared for all tests) or at the start of this test (eg via `DataTestUtil.createVolumeAndBucket`).
##########
hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/replication/TestPerVolumePushReplication.java:
##########
@@ -0,0 +1,526 @@
+/*
+ * 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.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.OzoneTestHelper.waitForReplicaCount;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.BooleanSupplier;
+import java.util.function.ToLongFunction;
+import java.util.stream.Collectors;
+import org.apache.hadoop.hdds.HddsConfigKeys;
+import org.apache.hadoop.hdds.client.BlockID;
+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.protocol.datanode.proto.ContainerProtos;
+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.HddsDatanodeService;
+import org.apache.hadoop.ozone.MiniOzoneCluster;
+import org.apache.hadoop.ozone.OzoneConfigKeys;
+import org.apache.hadoop.ozone.DataTestUtil;
+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.ContainerTestHelper;
+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;
+import org.junit.jupiter.api.parallel.Execution;
+import org.junit.jupiter.api.parallel.ExecutionMode;
+import org.junit.jupiter.api.parallel.ResourceLock;
+
+/**
+ * Integration tests for per-volume push replication thread pools (HDDS-15412).
+ */
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
+@Execution(ExecutionMode.SAME_THREAD)
+@ResourceLock("MiniOzoneCluster")
+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 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;
+
+ @BeforeAll
+ void setUp() throws Exception {
+ OzoneConfiguration conf = createSharedConfig();
+ cluster = newCluster(conf, DATANODE_COUNT);
+ cluster.waitForClusterToBeReady();
+ clientFactory = new XceiverClientManager(conf);
+ }
+
+ @AfterAll
+ void tearDown() {
+ IOUtils.closeQuietly(clientFactory, cluster);
+ }
+
+ @Order(1)
+ @Test
+ void testPushAndScmReplicationWithPerVolumeEnabled() throws Exception {
+ try (OzoneClient client = cluster.newClient()) {
+ HddsDatanodeService sourceDn = selectHealthyDatanode(0);
+ DatanodeDetails source = sourceDn.getDatanodeDetails();
+ DatanodeDetails target = selectOtherHealthyNode(source);
+ long containerId = createClosedContainer(clientFactory, source, 0L);
+
+ ReplicateContainerCommand cmd =
+ ReplicateContainerCommand.toTarget(containerId, target);
+ queuePushAndWait(cluster, cmd, source, ReplicationSupervisor::getReplicationSuccessCount);
+ assertNotNull(getContainer(cluster, target, containerId));
+ assertVolumePools(sourceDn, DATA_VOLUMES, 1);
+
+ OzoneBucket bucket = DataTestUtil.createVolumeAndBucket(client, VOLUME, BUCKET);
+ 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);
+
+ ContainerManager containerManager = cluster.getStorageContainerManager().getContainerManager();
+ Set<ContainerReplica> replicas =
+ containerManager.getContainerReplicas(ContainerID.valueOf(scmContainerId));
+ DatanodeDetails replicaDn = replicas.iterator().next().getDatanodeDetails();
+ cluster.shutdownHddsDatanode(replicaDn);
+ waitForReplicaCount(scmContainerId, 3, cluster);
+ }
+ }
+
+ @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, 0L);
+ long containerOnVol1 = findOrCreateContainerOnVolume(
+ cluster, clientFactory, source, vol1, 0L);
+
+ assertVolumePools(sourceDn, DATA_VOLUMES, 1);
+
+ triggerAndWaitForVolumeFailure(volSet, vol0);
+ waitForVolumePoolState(sourceDn, vol0, vol1);
+
+ ReplicateContainerCommand cmd =
+ ReplicateContainerCommand.toTarget(containerOnVol1, target);
+ queuePushAndWait(cluster, cmd, source, ReplicationSupervisor::getReplicationSuccessCount);
+ assertNotNull(getContainer(cluster, target, containerOnVol1));
Review Comment:
`getContainer(...)` never returns null (it throws `AssertionError`), so `assertNotNull(getContainer(...))` is redundant and can be misleading. Either (a) drop `assertNotNull(...)` and just call `getContainer(...)`, or (b) change `getContainer(...)` to return null/`Optional` and keep the explicit assertion.
##########
hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/replication/TestPerVolumePushReplication.java:
##########
@@ -0,0 +1,526 @@
+/*
+ * 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.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.OzoneTestHelper.waitForReplicaCount;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.BooleanSupplier;
+import java.util.function.ToLongFunction;
+import java.util.stream.Collectors;
+import org.apache.hadoop.hdds.HddsConfigKeys;
+import org.apache.hadoop.hdds.client.BlockID;
+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.protocol.datanode.proto.ContainerProtos;
+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.HddsDatanodeService;
+import org.apache.hadoop.ozone.MiniOzoneCluster;
+import org.apache.hadoop.ozone.OzoneConfigKeys;
+import org.apache.hadoop.ozone.DataTestUtil;
+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.ContainerTestHelper;
+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;
+import org.junit.jupiter.api.parallel.Execution;
+import org.junit.jupiter.api.parallel.ExecutionMode;
+import org.junit.jupiter.api.parallel.ResourceLock;
+
+/**
+ * Integration tests for per-volume push replication thread pools (HDDS-15412).
+ */
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
+@Execution(ExecutionMode.SAME_THREAD)
+@ResourceLock("MiniOzoneCluster")
+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 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;
+
+ @BeforeAll
+ void setUp() throws Exception {
+ OzoneConfiguration conf = createSharedConfig();
+ cluster = newCluster(conf, DATANODE_COUNT);
+ cluster.waitForClusterToBeReady();
+ clientFactory = new XceiverClientManager(conf);
+ }
+
+ @AfterAll
+ void tearDown() {
+ IOUtils.closeQuietly(clientFactory, cluster);
+ }
+
+ @Order(1)
+ @Test
+ void testPushAndScmReplicationWithPerVolumeEnabled() throws Exception {
+ try (OzoneClient client = cluster.newClient()) {
+ HddsDatanodeService sourceDn = selectHealthyDatanode(0);
+ DatanodeDetails source = sourceDn.getDatanodeDetails();
+ DatanodeDetails target = selectOtherHealthyNode(source);
+ long containerId = createClosedContainer(clientFactory, source, 0L);
+
+ ReplicateContainerCommand cmd =
+ ReplicateContainerCommand.toTarget(containerId, target);
+ queuePushAndWait(cluster, cmd, source, ReplicationSupervisor::getReplicationSuccessCount);
+ assertNotNull(getContainer(cluster, target, containerId));
+ assertVolumePools(sourceDn, DATA_VOLUMES, 1);
+
+ OzoneBucket bucket = DataTestUtil.createVolumeAndBucket(client, VOLUME, BUCKET);
+ 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);
+
+ ContainerManager containerManager = cluster.getStorageContainerManager().getContainerManager();
+ Set<ContainerReplica> replicas =
+ containerManager.getContainerReplicas(ContainerID.valueOf(scmContainerId));
+ DatanodeDetails replicaDn = replicas.iterator().next().getDatanodeDetails();
+ cluster.shutdownHddsDatanode(replicaDn);
+ waitForReplicaCount(scmContainerId, 3, cluster);
+ }
+ }
+
+ @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, 0L);
+ long containerOnVol1 = findOrCreateContainerOnVolume(
+ cluster, clientFactory, source, vol1, 0L);
+
+ assertVolumePools(sourceDn, DATA_VOLUMES, 1);
+
+ triggerAndWaitForVolumeFailure(volSet, vol0);
+ waitForVolumePoolState(sourceDn, vol0, vol1);
+
+ ReplicateContainerCommand cmd =
+ ReplicateContainerCommand.toTarget(containerOnVol1, target);
+ queuePushAndWait(cluster, cmd, source, ReplicationSupervisor::getReplicationSuccessCount);
+ assertNotNull(getContainer(cluster, target, containerOnVol1));
+ assertEquals(1, volSet.getFailedVolumesList().size());
+
+ // Task routes via global pool fallback (HDDS-15327); replication fails on bad volume.
+ ReplicateContainerCommand failedVolCmd =
+ ReplicateContainerCommand.toTarget(containerOnVol0, target);
+ ReplicationSupervisor supervisor =
+ sourceDn.getDatanodeStateMachine().getSupervisor();
+ long previousFailures = supervisor.getReplicationFailureCount();
+ queuePushAndWait(cluster, failedVolCmd, source,
+ ReplicationSupervisor::getReplicationFailureCount);
+ assertTrue(supervisor.getReplicationFailureCount() >= previousFailures + 1);
+
+ DatanodeTestUtils.restoreBadVolume(vol0);
+ }
+
+ @Order(3)
+ @Test
+ void testDecommissionWithPerVolumePools() throws Exception {
+ try (OzoneClient client = cluster.newClient();
+ ContainerOperationClient scmClient = new ContainerOperationClient(cluster.getConf())) {
+ StorageContainerManager scm = cluster.getStorageContainerManager();
+ NodeManager nm = scm.getScmNodeManager();
+ ContainerManager cm = scm.getContainerManager();
+ PipelineManager pm = scm.getPipelineManager();
+
+ OzoneBucket bucket = client.getObjectStore().getVolume(VOLUME).getBucket(BUCKET);
+ 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);
+
+ scmClient.decommissionNodes(singletonList(getDNHostAndPort(toDecommission)), false);
+ waitForDnToReachOpState(nm, toDecommission, DECOMMISSIONED);
+
+ 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(StandardCharsets.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) {
+ 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 void assertVolumePools(HddsDatanodeService dn,
+ int expectedVolumeCount, int expectedPoolSize) {
+ VolumeReplicationThreadPools pools =
+ dn.getDatanodeStateMachine().getSupervisor().getVolumeReplicationThreadPools();
+ assertNotNull(pools);
+ List<? extends StorageVolume> volumes =
+ dn.getDatanodeStateMachine().getContainer().getVolumeSet().getVolumesList();
+ assertEquals(expectedVolumeCount, volumes.size());
+ for (StorageVolume volume : volumes) {
+ String volumeRoot = volume.getStorageDir().getPath();
+ assertTrue(pools.hasPool(volumeRoot), "missing pool for " + volumeRoot);
+ assertEquals(expectedPoolSize, pools.getPoolSize(volumeRoot));
+ }
+ }
+
+ 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((BooleanSupplier) () -> {
+ VolumeReplicationThreadPools pools =
+ sourceDn.getDatanodeStateMachine().getSupervisor().getVolumeReplicationThreadPools();
+ return pools != null
+ && !pools.hasPool(failedPath)
+ && pools.hasPool(healthyPath);
+ }, 100, 60000);
+ VolumeReplicationThreadPools pools =
+ sourceDn.getDatanodeStateMachine().getSupervisor().getVolumeReplicationThreadPools();
+ assertNotNull(pools);
+ assertFalse(pools.hasPool(failedPath));
+ assertTrue(pools.hasPool(healthyPath));
+ }
+
+ private static void queuePushAndWait(MiniOzoneCluster cluster,
+ ReplicateContainerCommand cmd, DatanodeDetails source,
+ ToLongFunction<ReplicationSupervisor> counter)
+ throws IOException, InterruptedException, TimeoutException {
+ DatanodeStateMachine stateMachine = cluster.getHddsDatanode(source).getDatanodeStateMachine();
+ ReplicationSupervisor supervisor = stateMachine.getSupervisor();
+ long previousCount = counter.applyAsLong(supervisor);
+ long targetCount = previousCount + 1;
+ StateContext context = stateMachine.getContext();
+ context.getTermOfLeaderSCM().ifPresent(cmd::setTerm);
+ context.addCommand(cmd);
+ GenericTestUtils.waitFor((BooleanSupplier) () ->
+ counter.applyAsLong(supervisor) >= targetCount, 100, 30000);
Review Comment:
Waiting on global success/failure counters can produce false positives in a shared cluster: unrelated background replication activity may increment the counter and satisfy the condition before this specific `cmd` completes. This can make the tests flaky (eg `getContainer(...)` may fail right after `queuePushAndWait`). Prefer waiting on a command/container-specific condition, such as (for success) polling until the target datanode actually has the replicated container, and (for expected failure) asserting the container is still absent on the target after a timeout while also checking the failure counter increased.
##########
hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/replication/TestPerVolumePushReplication.java:
##########
@@ -0,0 +1,526 @@
+/*
+ * 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.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.OzoneTestHelper.waitForReplicaCount;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.BooleanSupplier;
+import java.util.function.ToLongFunction;
+import java.util.stream.Collectors;
+import org.apache.hadoop.hdds.HddsConfigKeys;
+import org.apache.hadoop.hdds.client.BlockID;
+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.protocol.datanode.proto.ContainerProtos;
+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.HddsDatanodeService;
+import org.apache.hadoop.ozone.MiniOzoneCluster;
+import org.apache.hadoop.ozone.OzoneConfigKeys;
+import org.apache.hadoop.ozone.DataTestUtil;
+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.ContainerTestHelper;
+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;
+import org.junit.jupiter.api.parallel.Execution;
+import org.junit.jupiter.api.parallel.ExecutionMode;
+import org.junit.jupiter.api.parallel.ResourceLock;
+
+/**
+ * Integration tests for per-volume push replication thread pools (HDDS-15412).
+ */
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
+@Execution(ExecutionMode.SAME_THREAD)
+@ResourceLock("MiniOzoneCluster")
+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 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;
+
+ @BeforeAll
+ void setUp() throws Exception {
+ OzoneConfiguration conf = createSharedConfig();
+ cluster = newCluster(conf, DATANODE_COUNT);
+ cluster.waitForClusterToBeReady();
+ clientFactory = new XceiverClientManager(conf);
+ }
+
+ @AfterAll
+ void tearDown() {
+ IOUtils.closeQuietly(clientFactory, cluster);
+ }
+
+ @Order(1)
+ @Test
+ void testPushAndScmReplicationWithPerVolumeEnabled() throws Exception {
+ try (OzoneClient client = cluster.newClient()) {
+ HddsDatanodeService sourceDn = selectHealthyDatanode(0);
+ DatanodeDetails source = sourceDn.getDatanodeDetails();
+ DatanodeDetails target = selectOtherHealthyNode(source);
+ long containerId = createClosedContainer(clientFactory, source, 0L);
+
+ ReplicateContainerCommand cmd =
+ ReplicateContainerCommand.toTarget(containerId, target);
+ queuePushAndWait(cluster, cmd, source, ReplicationSupervisor::getReplicationSuccessCount);
+ assertNotNull(getContainer(cluster, target, containerId));
+ assertVolumePools(sourceDn, DATA_VOLUMES, 1);
+
+ OzoneBucket bucket = DataTestUtil.createVolumeAndBucket(client, VOLUME, BUCKET);
+ 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);
+
+ ContainerManager containerManager = cluster.getStorageContainerManager().getContainerManager();
+ Set<ContainerReplica> replicas =
+ containerManager.getContainerReplicas(ContainerID.valueOf(scmContainerId));
+ DatanodeDetails replicaDn = replicas.iterator().next().getDatanodeDetails();
+ cluster.shutdownHddsDatanode(replicaDn);
+ waitForReplicaCount(scmContainerId, 3, cluster);
+ }
+ }
+
+ @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, 0L);
+ long containerOnVol1 = findOrCreateContainerOnVolume(
+ cluster, clientFactory, source, vol1, 0L);
+
+ assertVolumePools(sourceDn, DATA_VOLUMES, 1);
+
+ triggerAndWaitForVolumeFailure(volSet, vol0);
+ waitForVolumePoolState(sourceDn, vol0, vol1);
+
+ ReplicateContainerCommand cmd =
+ ReplicateContainerCommand.toTarget(containerOnVol1, target);
+ queuePushAndWait(cluster, cmd, source, ReplicationSupervisor::getReplicationSuccessCount);
+ assertNotNull(getContainer(cluster, target, containerOnVol1));
+ assertEquals(1, volSet.getFailedVolumesList().size());
+
+ // Task routes via global pool fallback (HDDS-15327); replication fails on bad volume.
+ ReplicateContainerCommand failedVolCmd =
+ ReplicateContainerCommand.toTarget(containerOnVol0, target);
+ ReplicationSupervisor supervisor =
+ sourceDn.getDatanodeStateMachine().getSupervisor();
+ long previousFailures = supervisor.getReplicationFailureCount();
+ queuePushAndWait(cluster, failedVolCmd, source,
+ ReplicationSupervisor::getReplicationFailureCount);
+ assertTrue(supervisor.getReplicationFailureCount() >= previousFailures + 1);
+
+ DatanodeTestUtils.restoreBadVolume(vol0);
+ }
+
+ @Order(3)
+ @Test
+ void testDecommissionWithPerVolumePools() throws Exception {
+ try (OzoneClient client = cluster.newClient();
+ ContainerOperationClient scmClient = new ContainerOperationClient(cluster.getConf())) {
+ StorageContainerManager scm = cluster.getStorageContainerManager();
+ NodeManager nm = scm.getScmNodeManager();
+ ContainerManager cm = scm.getContainerManager();
+ PipelineManager pm = scm.getPipelineManager();
+
+ OzoneBucket bucket = client.getObjectStore().getVolume(VOLUME).getBucket(BUCKET);
+ 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);
+
+ scmClient.decommissionNodes(singletonList(getDNHostAndPort(toDecommission)), false);
+ waitForDnToReachOpState(nm, toDecommission, DECOMMISSIONED);
+
+ 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(StandardCharsets.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) {
+ 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 void assertVolumePools(HddsDatanodeService dn,
+ int expectedVolumeCount, int expectedPoolSize) {
+ VolumeReplicationThreadPools pools =
+ dn.getDatanodeStateMachine().getSupervisor().getVolumeReplicationThreadPools();
+ assertNotNull(pools);
+ List<? extends StorageVolume> volumes =
+ dn.getDatanodeStateMachine().getContainer().getVolumeSet().getVolumesList();
+ assertEquals(expectedVolumeCount, volumes.size());
+ for (StorageVolume volume : volumes) {
+ String volumeRoot = volume.getStorageDir().getPath();
+ assertTrue(pools.hasPool(volumeRoot), "missing pool for " + volumeRoot);
+ assertEquals(expectedPoolSize, pools.getPoolSize(volumeRoot));
+ }
+ }
+
+ 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((BooleanSupplier) () -> {
+ VolumeReplicationThreadPools pools =
+ sourceDn.getDatanodeStateMachine().getSupervisor().getVolumeReplicationThreadPools();
+ return pools != null
+ && !pools.hasPool(failedPath)
+ && pools.hasPool(healthyPath);
+ }, 100, 60000);
+ VolumeReplicationThreadPools pools =
+ sourceDn.getDatanodeStateMachine().getSupervisor().getVolumeReplicationThreadPools();
+ assertNotNull(pools);
+ assertFalse(pools.hasPool(failedPath));
+ assertTrue(pools.hasPool(healthyPath));
+ }
+
+ private static void queuePushAndWait(MiniOzoneCluster cluster,
+ ReplicateContainerCommand cmd, DatanodeDetails source,
+ ToLongFunction<ReplicationSupervisor> counter)
+ throws IOException, InterruptedException, TimeoutException {
+ DatanodeStateMachine stateMachine = cluster.getHddsDatanode(source).getDatanodeStateMachine();
+ ReplicationSupervisor supervisor = stateMachine.getSupervisor();
+ long previousCount = counter.applyAsLong(supervisor);
+ long targetCount = previousCount + 1;
+ StateContext context = stateMachine.getContext();
+ context.getTermOfLeaderSCM().ifPresent(cmd::setTerm);
+ context.addCommand(cmd);
+ GenericTestUtils.waitFor((BooleanSupplier) () ->
+ counter.applyAsLong(supervisor) >= targetCount, 100, 30000);
+ }
+
+ private static long findOrCreateContainerOnVolume(MiniOzoneCluster cluster,
+ XceiverClientFactory clientFactory, DatanodeDetails dn, HddsVolume targetVolume,
+ long dataSize) throws Exception {
+ for (int attempt = 0; attempt < 30; attempt++) {
+ long containerId = createClosedContainer(clientFactory, dn, dataSize);
+ 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, long dataSize) throws Exception {
+ long containerId = CONTAINER_ID.incrementAndGet();
+ try (XceiverClientSpi client = clientFactory.acquireClient(createPipeline(singleton(dn)))) {
+ if (dataSize <= 0) {
+ createContainer(client, containerId, null, CLOSED, 0);
+ return containerId;
+ }
+ createContainer(client, containerId, null);
+ int chunkSize = 1024 * 1024;
+ long totalBytesWritten = 0;
+ while (totalBytesWritten < dataSize) {
+ BlockID blockId = ContainerTestHelper.getTestBlockID(containerId);
+ long remainingBytes = dataSize - totalBytesWritten;
+ int currentChunkSize = (int) Math.min(chunkSize, remainingBytes);
+ ContainerProtos.ContainerCommandRequestProto writeChunkRequest =
+ ContainerTestHelper.getWriteChunkRequest(
+ createPipeline(singleton(dn)), blockId, currentChunkSize);
+ client.sendCommand(writeChunkRequest);
+ ContainerProtos.ContainerCommandRequestProto putBlockRequest =
+ ContainerTestHelper.getPutBlockRequest(writeChunkRequest);
+ client.sendCommand(putBlockRequest);
+ totalBytesWritten += currentChunkSize;
+ }
+ ContainerProtos.CloseContainerRequestProto closeRequest =
+ ContainerProtos.CloseContainerRequestProto.newBuilder().build();
+ ContainerProtos.ContainerCommandRequestProto closeContainerRequest =
+ ContainerProtos.ContainerCommandRequestProto.newBuilder()
+ .setCmdType(ContainerProtos.Type.CloseContainer)
+ .setContainerID(containerId)
+ .setCloseContainer(closeRequest)
+ .setDatanodeUuid(dn.getUuidString())
+ .build();
+ client.sendCommand(closeContainerRequest);
+ 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();
+ ReplicationServer.ReplicationConfig clusterReplicationConfig =
+ conf.getObject(ReplicationServer.ReplicationConfig.class);
+ return MiniOzoneCluster.newBuilder(conf)
+ .setNumDatanodes(numDatanodes)
+ .setDatanodeFactory(baseConf -> {
+ OzoneConfiguration dnConf = uniformFactory.apply(baseConf);
+ ReplicationServer.ReplicationConfig dnReplicationConfig =
+ dnConf.getObject(ReplicationServer.ReplicationConfig.class);
+ dnReplicationConfig.setPerVolumeEnabled(
+ clusterReplicationConfig.isPerVolumeEnabled());
+ dnReplicationConfig.setPerVolumeStreamsLimit(
+ clusterReplicationConfig.getPerVolumeStreamsLimit());
+ dnConf.setFromObject(dnReplicationConfig);
+ return dnConf;
+ })
+ .build();
+ }
+
+ private static OzoneConfiguration createSharedConfig() {
+ OzoneConfiguration conf = createDecommissionConfig(true, 1);
+ configureVolumeFailureDetection(conf);
+ return conf;
+ }
+
+ private static OzoneConfiguration createPerVolumeConfig(boolean perVolumeEnabled,
+ int streamsLimit) {
+ 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.setStorageSize(ScmConfigKeys.OZONE_SCM_CONTAINER_SIZE, 5, StorageUnit.MB);
+ conf.setStorageSize(OzoneConfigKeys.OZONE_SCM_BLOCK_SIZE, 1, StorageUnit.MB);
+
+ ReplicationManagerConfiguration repConf =
+ conf.getObject(ReplicationManagerConfiguration.class);
+ repConf.setInterval(Duration.ofSeconds(1));
+ repConf.setUnderReplicatedInterval(Duration.ofSeconds(1));
+ conf.setFromObject(repConf);
+
+ ReplicationServer.ReplicationConfig replicationConfig =
+ conf.getObject(ReplicationServer.ReplicationConfig.class);
+ replicationConfig.setPerVolumeEnabled(perVolumeEnabled);
+ replicationConfig.setPerVolumeStreamsLimit(streamsLimit);
+ conf.setFromObject(replicationConfig);
+ return conf;
+ }
+
+ private static OzoneConfiguration createDecommissionConfig(boolean perVolumeEnabled,
+ int streamsLimit) {
+ OzoneConfiguration conf = createPerVolumeConfig(perVolumeEnabled, streamsLimit);
+ 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.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);
+ conf.setTimeDuration(
+ ScmConfigKeys.OZONE_SCM_EXPIRED_CONTAINER_REPLICA_OP_SCRUB_INTERVAL, 1, TimeUnit.SECONDS);
+
+ ReplicationManagerConfiguration repConf =
+ conf.getObject(ReplicationManagerConfiguration.class);
+ repConf.setOverReplicatedInterval(Duration.ofSeconds(1));
+ conf.setFromObject(repConf);
+ return conf;
+ }
+
+ private static void configureVolumeFailureDetection(OzoneConfiguration conf) {
+ DatanodeConfiguration dnConf = conf.getObject(DatanodeConfiguration.class);
+ dnConf.setFailedDataVolumesTolerated(1);
+ dnConf.setDiskCheckMinGap(Duration.ofSeconds(0));
+ dnConf.setPeriodicDiskCheckIntervalMinutes(1);
+ conf.setFromObject(dnConf);
+ }
+
+ private static void triggerAndWaitForVolumeFailure(MutableVolumeSet volSet,
+ StorageVolume volume) throws Exception {
+ DatanodeTestUtils.simulateBadVolume(volume);
+ volSet.checkVolumeAsync(volume);
+ GenericTestUtils.waitFor(
+ (BooleanSupplier) () -> volSet.getFailedVolumesList().size() == 1,
+ 100, 60000);
+ }
+
+ 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(StandardCharsets.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 {
+ GenericTestUtils.waitFor(
+ (BooleanSupplier) () -> getContainerReplicas(cm, container).size() == count,
+ 200, 60000);
+ }
+
+ private static Set<ContainerReplica> getContainerReplicas(ContainerManager cm,
+ ContainerInfo container) {
+ try {
+ return cm.getContainerReplicas(container.containerID());
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
Review Comment:
The `waitFor(...)` predicate can throw via `getContainerReplicas` (wrapped as `RuntimeException`), which will abort the wait immediately rather than retrying. If `cm.getContainerReplicas(...)` can transiently fail in this mini-cluster flow, this becomes a source of test flakiness. Consider handling exceptions inside the wait predicate (eg treat exceptions as 'not ready yet' and return false) or have `getContainerReplicas` return an empty set on transient errors and let the wait continue.
##########
hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/replication/TestPerVolumePushReplication.java:
##########
@@ -0,0 +1,526 @@
+/*
+ * 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.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.OzoneTestHelper.waitForReplicaCount;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.BooleanSupplier;
+import java.util.function.ToLongFunction;
+import java.util.stream.Collectors;
+import org.apache.hadoop.hdds.HddsConfigKeys;
+import org.apache.hadoop.hdds.client.BlockID;
+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.protocol.datanode.proto.ContainerProtos;
+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.HddsDatanodeService;
+import org.apache.hadoop.ozone.MiniOzoneCluster;
+import org.apache.hadoop.ozone.OzoneConfigKeys;
+import org.apache.hadoop.ozone.DataTestUtil;
+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.ContainerTestHelper;
+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;
+import org.junit.jupiter.api.parallel.Execution;
+import org.junit.jupiter.api.parallel.ExecutionMode;
+import org.junit.jupiter.api.parallel.ResourceLock;
+
+/**
+ * Integration tests for per-volume push replication thread pools (HDDS-15412).
+ */
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
+@Execution(ExecutionMode.SAME_THREAD)
+@ResourceLock("MiniOzoneCluster")
+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 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;
+
+ @BeforeAll
+ void setUp() throws Exception {
+ OzoneConfiguration conf = createSharedConfig();
+ cluster = newCluster(conf, DATANODE_COUNT);
+ cluster.waitForClusterToBeReady();
+ clientFactory = new XceiverClientManager(conf);
+ }
+
+ @AfterAll
+ void tearDown() {
+ IOUtils.closeQuietly(clientFactory, cluster);
+ }
+
+ @Order(1)
+ @Test
+ void testPushAndScmReplicationWithPerVolumeEnabled() throws Exception {
+ try (OzoneClient client = cluster.newClient()) {
+ HddsDatanodeService sourceDn = selectHealthyDatanode(0);
+ DatanodeDetails source = sourceDn.getDatanodeDetails();
+ DatanodeDetails target = selectOtherHealthyNode(source);
+ long containerId = createClosedContainer(clientFactory, source, 0L);
+
+ ReplicateContainerCommand cmd =
+ ReplicateContainerCommand.toTarget(containerId, target);
+ queuePushAndWait(cluster, cmd, source, ReplicationSupervisor::getReplicationSuccessCount);
+ assertNotNull(getContainer(cluster, target, containerId));
Review Comment:
`getContainer(...)` never returns null (it throws `AssertionError`), so `assertNotNull(getContainer(...))` is redundant and can be misleading. Either (a) drop `assertNotNull(...)` and just call `getContainer(...)`, or (b) change `getContainer(...)` to return null/`Optional` and keep the explicit assertion.
##########
hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/replication/TestPerVolumePushReplication.java:
##########
@@ -0,0 +1,526 @@
+/*
+ * 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.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.OzoneTestHelper.waitForReplicaCount;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.BooleanSupplier;
+import java.util.function.ToLongFunction;
+import java.util.stream.Collectors;
+import org.apache.hadoop.hdds.HddsConfigKeys;
+import org.apache.hadoop.hdds.client.BlockID;
+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.protocol.datanode.proto.ContainerProtos;
+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.HddsDatanodeService;
+import org.apache.hadoop.ozone.MiniOzoneCluster;
+import org.apache.hadoop.ozone.OzoneConfigKeys;
+import org.apache.hadoop.ozone.DataTestUtil;
+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.ContainerTestHelper;
+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;
+import org.junit.jupiter.api.parallel.Execution;
+import org.junit.jupiter.api.parallel.ExecutionMode;
+import org.junit.jupiter.api.parallel.ResourceLock;
+
+/**
+ * Integration tests for per-volume push replication thread pools (HDDS-15412).
+ */
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
+@Execution(ExecutionMode.SAME_THREAD)
+@ResourceLock("MiniOzoneCluster")
+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 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;
+
+ @BeforeAll
+ void setUp() throws Exception {
+ OzoneConfiguration conf = createSharedConfig();
+ cluster = newCluster(conf, DATANODE_COUNT);
+ cluster.waitForClusterToBeReady();
+ clientFactory = new XceiverClientManager(conf);
+ }
+
+ @AfterAll
+ void tearDown() {
+ IOUtils.closeQuietly(clientFactory, cluster);
+ }
+
+ @Order(1)
+ @Test
+ void testPushAndScmReplicationWithPerVolumeEnabled() throws Exception {
+ try (OzoneClient client = cluster.newClient()) {
+ HddsDatanodeService sourceDn = selectHealthyDatanode(0);
+ DatanodeDetails source = sourceDn.getDatanodeDetails();
+ DatanodeDetails target = selectOtherHealthyNode(source);
+ long containerId = createClosedContainer(clientFactory, source, 0L);
+
+ ReplicateContainerCommand cmd =
+ ReplicateContainerCommand.toTarget(containerId, target);
+ queuePushAndWait(cluster, cmd, source, ReplicationSupervisor::getReplicationSuccessCount);
+ assertNotNull(getContainer(cluster, target, containerId));
+ assertVolumePools(sourceDn, DATA_VOLUMES, 1);
+
+ OzoneBucket bucket = DataTestUtil.createVolumeAndBucket(client, VOLUME, BUCKET);
+ 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);
+
+ ContainerManager containerManager = cluster.getStorageContainerManager().getContainerManager();
+ Set<ContainerReplica> replicas =
+ containerManager.getContainerReplicas(ContainerID.valueOf(scmContainerId));
+ DatanodeDetails replicaDn = replicas.iterator().next().getDatanodeDetails();
+ cluster.shutdownHddsDatanode(replicaDn);
+ waitForReplicaCount(scmContainerId, 3, cluster);
+ }
+ }
+
+ @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, 0L);
+ long containerOnVol1 = findOrCreateContainerOnVolume(
+ cluster, clientFactory, source, vol1, 0L);
+
+ assertVolumePools(sourceDn, DATA_VOLUMES, 1);
+
+ triggerAndWaitForVolumeFailure(volSet, vol0);
+ waitForVolumePoolState(sourceDn, vol0, vol1);
+
+ ReplicateContainerCommand cmd =
+ ReplicateContainerCommand.toTarget(containerOnVol1, target);
+ queuePushAndWait(cluster, cmd, source, ReplicationSupervisor::getReplicationSuccessCount);
+ assertNotNull(getContainer(cluster, target, containerOnVol1));
+ assertEquals(1, volSet.getFailedVolumesList().size());
+
+ // Task routes via global pool fallback (HDDS-15327); replication fails on bad volume.
+ ReplicateContainerCommand failedVolCmd =
+ ReplicateContainerCommand.toTarget(containerOnVol0, target);
+ ReplicationSupervisor supervisor =
+ sourceDn.getDatanodeStateMachine().getSupervisor();
+ long previousFailures = supervisor.getReplicationFailureCount();
+ queuePushAndWait(cluster, failedVolCmd, source,
+ ReplicationSupervisor::getReplicationFailureCount);
+ assertTrue(supervisor.getReplicationFailureCount() >= previousFailures + 1);
+
+ DatanodeTestUtils.restoreBadVolume(vol0);
+ }
+
+ @Order(3)
+ @Test
+ void testDecommissionWithPerVolumePools() throws Exception {
+ try (OzoneClient client = cluster.newClient();
+ ContainerOperationClient scmClient = new ContainerOperationClient(cluster.getConf())) {
+ StorageContainerManager scm = cluster.getStorageContainerManager();
+ NodeManager nm = scm.getScmNodeManager();
+ ContainerManager cm = scm.getContainerManager();
+ PipelineManager pm = scm.getPipelineManager();
+
+ OzoneBucket bucket = client.getObjectStore().getVolume(VOLUME).getBucket(BUCKET);
+ 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);
+
+ scmClient.decommissionNodes(singletonList(getDNHostAndPort(toDecommission)), false);
+ waitForDnToReachOpState(nm, toDecommission, DECOMMISSIONED);
+
+ 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(StandardCharsets.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) {
+ 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 void assertVolumePools(HddsDatanodeService dn,
+ int expectedVolumeCount, int expectedPoolSize) {
+ VolumeReplicationThreadPools pools =
+ dn.getDatanodeStateMachine().getSupervisor().getVolumeReplicationThreadPools();
+ assertNotNull(pools);
+ List<? extends StorageVolume> volumes =
+ dn.getDatanodeStateMachine().getContainer().getVolumeSet().getVolumesList();
+ assertEquals(expectedVolumeCount, volumes.size());
+ for (StorageVolume volume : volumes) {
+ String volumeRoot = volume.getStorageDir().getPath();
+ assertTrue(pools.hasPool(volumeRoot), "missing pool for " + volumeRoot);
+ assertEquals(expectedPoolSize, pools.getPoolSize(volumeRoot));
+ }
+ }
+
+ 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((BooleanSupplier) () -> {
+ VolumeReplicationThreadPools pools =
+ sourceDn.getDatanodeStateMachine().getSupervisor().getVolumeReplicationThreadPools();
+ return pools != null
+ && !pools.hasPool(failedPath)
+ && pools.hasPool(healthyPath);
+ }, 100, 60000);
+ VolumeReplicationThreadPools pools =
+ sourceDn.getDatanodeStateMachine().getSupervisor().getVolumeReplicationThreadPools();
+ assertNotNull(pools);
+ assertFalse(pools.hasPool(failedPath));
+ assertTrue(pools.hasPool(healthyPath));
+ }
+
+ private static void queuePushAndWait(MiniOzoneCluster cluster,
+ ReplicateContainerCommand cmd, DatanodeDetails source,
+ ToLongFunction<ReplicationSupervisor> counter)
+ throws IOException, InterruptedException, TimeoutException {
+ DatanodeStateMachine stateMachine = cluster.getHddsDatanode(source).getDatanodeStateMachine();
+ ReplicationSupervisor supervisor = stateMachine.getSupervisor();
+ long previousCount = counter.applyAsLong(supervisor);
+ long targetCount = previousCount + 1;
+ StateContext context = stateMachine.getContext();
+ context.getTermOfLeaderSCM().ifPresent(cmd::setTerm);
+ context.addCommand(cmd);
+ GenericTestUtils.waitFor((BooleanSupplier) () ->
+ counter.applyAsLong(supervisor) >= targetCount, 100, 30000);
+ }
+
+ private static long findOrCreateContainerOnVolume(MiniOzoneCluster cluster,
+ XceiverClientFactory clientFactory, DatanodeDetails dn, HddsVolume targetVolume,
+ long dataSize) throws Exception {
+ for (int attempt = 0; attempt < 30; attempt++) {
+ long containerId = createClosedContainer(clientFactory, dn, dataSize);
+ 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, long dataSize) throws Exception {
+ long containerId = CONTAINER_ID.incrementAndGet();
+ try (XceiverClientSpi client = clientFactory.acquireClient(createPipeline(singleton(dn)))) {
+ if (dataSize <= 0) {
+ createContainer(client, containerId, null, CLOSED, 0);
+ return containerId;
+ }
+ createContainer(client, containerId, null);
+ int chunkSize = 1024 * 1024;
+ long totalBytesWritten = 0;
+ while (totalBytesWritten < dataSize) {
+ BlockID blockId = ContainerTestHelper.getTestBlockID(containerId);
+ long remainingBytes = dataSize - totalBytesWritten;
+ int currentChunkSize = (int) Math.min(chunkSize, remainingBytes);
+ ContainerProtos.ContainerCommandRequestProto writeChunkRequest =
+ ContainerTestHelper.getWriteChunkRequest(
+ createPipeline(singleton(dn)), blockId, currentChunkSize);
+ client.sendCommand(writeChunkRequest);
+ ContainerProtos.ContainerCommandRequestProto putBlockRequest =
+ ContainerTestHelper.getPutBlockRequest(writeChunkRequest);
+ client.sendCommand(putBlockRequest);
+ totalBytesWritten += currentChunkSize;
+ }
+ ContainerProtos.CloseContainerRequestProto closeRequest =
+ ContainerProtos.CloseContainerRequestProto.newBuilder().build();
+ ContainerProtos.ContainerCommandRequestProto closeContainerRequest =
+ ContainerProtos.ContainerCommandRequestProto.newBuilder()
+ .setCmdType(ContainerProtos.Type.CloseContainer)
+ .setContainerID(containerId)
+ .setCloseContainer(closeRequest)
+ .setDatanodeUuid(dn.getUuidString())
+ .build();
+ client.sendCommand(closeContainerRequest);
+ 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;
+ }
Review Comment:
`getContainer(...)` never returns null (it throws `AssertionError`), so `assertNotNull(getContainer(...))` is redundant and can be misleading. Either (a) drop `assertNotNull(...)` and just call `getContainer(...)`, or (b) change `getContainer(...)` to return null/`Optional` and keep the explicit assertion.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-15802. Add integration tests for per-volume push replication [ozone]
Posted by "smengcl (via GitHub)" <gi...@apache.org>.
smengcl commented on PR #11146:
URL: https://github.com/apache/ozone/pull/11146#issuecomment-6042975963
Thanks @jojochuang for the patch. Thanks @rich7420 for the review.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-15802. Add integration tests for per-volume push replication [ozone]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #11146:
URL: https://github.com/apache/ozone/pull/11146#discussion_r4100737428
##########
hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/replication/TestPerVolumePushReplication.java:
##########
@@ -0,0 +1,547 @@
+/*
+ * 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.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);
+ 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);
+ 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);
+ 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);
+ }
Review Comment:
`restoreBadVolume` only restores directory writability; it does not return `vol0` from `failedVolumeMap` to the active volume set or recreate its replication pool. Consequently the ordered decommission test starts with one datanode still missing a volume, and flakes if that datanode is selected from the Ratis/EC pipeline intersection because `assertVolumePools` requires two pools. Restart this datanode after restoring the directory so the next test gets the fully restored cluster promised by the class contract.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-15802. Add integration tests for per-volume push replication [ozone]
Posted by "smengcl (via GitHub)" <gi...@apache.org>.
smengcl commented on PR #11146:
URL: https://github.com/apache/ozone/pull/11146#issuecomment-5825822711
Reviewed the current head. The four open comments from @rich7420 all hold, and so does the global-counter point (it reads as outdated but the underlying issue is still there). Below are only the issues **not** already raised in an existing comment. I have a patch for these that I can push to the branch if you like.
### 1. `waitForReplicaCount(scmContainerId, 3, cluster)` is satisfied on its first poll
`DeadNodeHandler` removes a datanode's replicas only once SCM declares the node DEAD (`ozone.scm.dead.node.interval` = 6s here). The wait runs milliseconds after `shutdownHddsDatanode()` returns, while SCM still has all three replicas registered, so the first poll succeeds and the method returns immediately.
The container is never observed under-replicated and re-replication never runs: this half of the test would pass identically with `ReplicationManager` disabled. Waiting for `DEAD` first, then for a three-replica set that excludes the stopped node, makes it deterministic and non-vacuous.
### 2. `replicas.iterator().next()` can throw `NoSuchElementException`
`waitForContainerClose()` polls datanode-side state only; SCM learns about replicas asynchronously via ICR. Immediately after it returns, `getContainerReplicas()` may hold fewer than three replicas, and in the worst case none, so `iterator().next()` throws instead of failing with a readable message. Settling on the replica count first fixes both this and (1).
### 3. `restoreBadVolume(vol0)` is not in a `finally`
`simulateBadVolume()` does `setWritable(false)`. If any assertion between the failure injection and the last line fails, the restore never runs: the decommission test then executes against a datanode with a permanently read-only volume, and `cluster.shutdown()` cannot delete its base dir (the exception is swallowed), leaking a read-only temp tree.
Worth noting separately that `restoreBadVolume` only flips the file permission. The volume stays in `getFailedVolumesList()` and its pool is not recreated, so it does not undo as much as the name suggests.
### 4. The decommission test's wait budget exceeds the 5m default JUnit timeout
The root `pom.xml` sets `junit.jupiter.execution.timeout.default = 5m` for every module, and `integration-test` overrides only `argLine`. `testDecommissionWithPerVolumePools` chains `waitForKeyContainer` x2 (60s each) + `waitForDnToReachOpState` (30s) + `waitForContainerReplicas` x2 (60s) + `waitForDnToReachHealthState` (30s) + `waitForContainerReplicas` x2 (60s) = 420s worst case, plus 40 key writes.
A slow CI agent aborts with a bare JUnit timeout naming no predicate, instead of the `waitFor` failure that would say which replica count never settled. `TestDecommissionAndMaintenance` uses `200, 30000` for the same waits.
### 5. `bucket` outlives the `OzoneClient` it came from
`setUp()` assigns the `bucket` field inside `try (OzoneClient client = cluster.newClient())`, so the client is closed before any test runs, while all three tests keep writing and reading keys through that bucket handle.
To be precise about severity: this does work today, because `RpcClient.close()` ends at `RPC.stopProxy`, and `ClientCache` refcounts the shared `ipc.Client` that the rest of the mini-cluster keeps alive, so the proxy stays usable. It is working by accident rather than by contract, and `XceiverClientManager.close()` has already invalidated the client cache and unregistered its metrics by then. `TestDecommissionAndMaintenance` holds `client` as a field and closes it in teardown; doing the same here costs three lines.
### 6. `UniformDatanodesFactory` silently resets every replication config key
This is the reason `newCluster()` needs its read/copy/write dance at all. `configureDatanodePorts()` does:
```java
conf.setFromObject(new ReplicationServer.ReplicationConfig().setPort(getFreePort()));
```
`setFromObject` writes every `@Config` field of a default-constructed object, so it resets `per.volume.enabled` to false, `per.volume.streams.limit` to 2, `streams.limit` to 10, `queue.limit` to 4096 and `outofservice.limit.factor` to 2.0 for **every** test using this factory.
The wrapper here restores only two of those six, so any other replication setting a future test puts in the cluster conf is still dropped, and the next person to hit this has to reinvent the same workaround. Setting just the port key in the factory would remove the need for the wrapper entirely. I have left this out of my patch because it is a change to shared test infrastructure rather than to this PR's test, but it seems worth a follow-up.
### 7. `@Execution(SAME_THREAD)` and `@ResourceLock("MiniOzoneCluster")` are no-ops
`pom.xml` already sets `parallel.mode.default = same_thread` and `parallel.mode.classes.default = same_thread` for all modules, so `SAME_THREAD` is the default. `@ResourceLock("MiniOzoneCluster")` is the only occurrence of that key in `integration-test`, so it can never contend with anything. Both read as a concurrency contract the build does not have.
### 8. Smaller items
- `assertTrue(supervisor.getReplicationFailureCount() >= previousFailures + 1)` prints only "expected true" and hides both counts; same for `assertEquals(1, volSet.getFailedVolumesList().size())` and the `hasPool` assertions. AssertJ (`assertThat(...).isGreaterThanOrEqualTo(...)`, `.hasSize(1)`) makes a CI failure diagnosable from the report alone, per HDDS-9951.
- `createSharedConfig` -> `createDecommissionConfig(boolean, int)` -> `createPerVolumeConfig(boolean, int)` is three methods for one configuration; both parameters only ever receive `(true, 1)`, and `ReplicationManagerConfiguration` is fetched and written back twice across two of them.
- The bare `getContainer(cluster, target, containerId)` calls after `queuePushAndWaitForContainer` are left over from the `assertNotNull` removal. The wait already confirmed presence via `hasContainer`, so these can only fire if the container vanished in the intervening microseconds.
- The file mixes the static import `UTF_8` with `StandardCharsets.UTF_8` for the same charset.
### On what the tests assert
Expanding on @rich7420's question at `testDecommissionWithPerVolumePools`, since it applies to the class as a whole: `assertVolumePools` and `waitForVolumePoolState` check pool existence and size, which `TestReplicationSupervisor#perVolumePoolSizeRespected` and its failed-volume test already cover without a cluster. Nothing currently asserts that a push was actually *dispatched* to a per-volume executor, so a push that ran entirely on the global pool would satisfy every assertion here.
Two assertions would make the cluster earn its place, and both are reachable from this package without touching production code:
- `pools.getExecutor(volumeRoot)` is package-private and returns the `ThreadPoolExecutor`, so `getCompletedTaskCount()` before/after a push proves the task ran on that volume's pool.
- For decommission specifically, SCM -> `SetNodeOperationalStateCommandHandler` -> `supervisor::nodeStateUpdated` -> `volumePools.setPoolSize(scaleOutOfServiceLimit(...))` is a path no unit test exercises end to end. Asserting the per-volume pools resize from 1 to `ceil(1 x 2.0)` = 2 once the node reaches `DECOMMISSIONED` is per-volume behaviour that the existing decommission test cannot catch.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-15802. Add integration tests for per-volume push replication [ozone]
Posted by "smengcl (via GitHub)" <gi...@apache.org>.
smengcl commented on PR #11146:
URL: https://github.com/apache/ozone/pull/11146#issuecomment-5826139472
Follow-up to my comment above: the patch is pushed to this branch as `66bf80e1221`. @jojochuang, I went ahead rather than waiting for a reply, so say the word if you would rather have it as a separate PR and I will move it.
Fixed: 1, 2, 3, 4, 5, 7, all of 8, and both assertions from the last section.
- (1) and (2): wait for the node to reach `DEAD`, then for a three-replica set that excludes the stopped node. The replica set is settled before a node is picked, so `iterator().next()` can no longer throw.
- (3): `restoreBadVolume` moved into a `finally`. The separate observation stands, it still only flips the file permission and does not clear the failed-volume list or recreate the pool.
- (4): every wait in the decommission test now routes through one 30s helper, worst case around 210s.
- (5): `client` is a field, closed in `@AfterAll` alongside the rest.
- (7): both annotations removed.
- (8): AssertJ throughout, the three config methods collapsed into one, the leftover no-op `getContainer(...)` calls dropped, single `UTF_8` import.
- Last section: per-volume dispatch is now asserted. Each push compares `ThreadPoolExecutor.getCompletedTaskCount()` on the pool of the volume holding the container, and the decommission test waits for the pools to resize to `scaleOutOfServiceLimit(...)`. Both reachable from this package, so no production change.
Not fixed: 6. `UniformDatanodesFactory.configureDatanodePorts` is shared test infrastructure and the fix changes behavior for every test using the factory, so it belongs in its own change. The `newCluster` wrapper stays for now.
One caveat on the patch: I verified it with compile and `checkstyle:check` only and could not run the three tests locally, so CI is the first real execution.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org