Copilot commented on code in PR #10703:
URL: https://github.com/apache/ozone/pull/10703#discussion_r3560870630
##########
hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationSupervisor.java:
##########
@@ -266,17 +302,68 @@ public void initCounters(AbstractReplicationTask task) {
}
private void addToQueue(AbstractReplicationTask task) {
- if (inFlight.add(task)) {
- if (task.getPriority() != ReplicationCommandPriority.LOW) {
- // Low priority tasks are not included in the replication queue sizes
- // returned to SCM in the heartbeat, so we only update the count for
- // priorities other than low.
- taskCounter.computeIfAbsent(task.getClass(),
- k -> new AtomicInteger()).incrementAndGet();
- }
- queuedCounter.get(task.getMetricName()).incrementAndGet();
- executor.execute(new TaskRunner(task));
+ if (!inFlight.add(task)) {
+ return;
+ }
+ if (task.getPriority() != ReplicationCommandPriority.LOW) {
+ taskCounter.computeIfAbsent(task.getClass(),
+ k -> new AtomicInteger()).incrementAndGet();
}
+ queuedCounter.get(task.getMetricName()).incrementAndGet();
+ ExecutorService taskExecutor = selectExecutor(task);
+ if (taskExecutor == null) {
+ LOG.warn("Rejected {} in ReplicationSupervisor: no replication handler "
+ + "thread pool available", task);
+ queuedCounter.get(task.getMetricName()).decrementAndGet();
+ inFlight.remove(task);
+ decrementTaskCounter(task);
+ return;
+ }
+ taskExecutor.execute(new TaskRunner(task));
Review Comment:
If task submission throws (eg executor shutdown/rejected), this method
currently leaves the task counted as in-flight/queued because the rollback only
happens when TaskRunner runs. Wrap the execute() call and revert
queuedCounter/inFlight/taskCounter on failure to avoid leaking metrics/state.
##########
hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/replication/TestPerVolumePushReplication.java:
##########
@@ -0,0 +1,471 @@
+/*
+ * 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.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.TestNodeUtil.getDNHostAndPort;
+import static
org.apache.hadoop.hdds.scm.node.TestNodeUtil.waitForDnToReachHealthState;
+import static
org.apache.hadoop.hdds.scm.node.TestNodeUtil.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.TestHelper.waitForContainerClose;
+import static org.apache.hadoop.ozone.container.TestHelper.waitForReplicaCount;
+import static org.apache.ozone.test.GenericTestUtils.waitFor;
+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.ToLongFunction;
+import java.util.stream.IntStream;
+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.ozone.HddsDatanodeService;
+import org.apache.hadoop.ozone.MiniOzoneCluster;
+import org.apache.hadoop.ozone.OzoneConfigKeys;
+import org.apache.hadoop.ozone.TestDataUtil;
+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.Test;
+
+/**
+ * Integration tests for per-volume push replication thread pools (HDDS-15412).
+ */
+class TestPerVolumePushReplication {
+
+ private static final AtomicLong CONTAINER_ID = new AtomicLong(1_000_000L);
+ private static final int DATA_VOLUMES = 2;
+ 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);
+
+ @Test
+ void testPushAndScmReplicationWithPerVolumeEnabled() throws Exception {
+ OzoneConfiguration conf = createPerVolumeConfig(true, 1);
+ try (MiniOzoneCluster cluster = newCluster(conf, 3);
+ XceiverClientFactory clientFactory = new XceiverClientManager(conf);
+ OzoneClient client = cluster.newClient()) {
+ cluster.waitForClusterToBeReady();
+
+ DatanodeDetails source =
cluster.getHddsDatanodes().get(0).getDatanodeDetails();
+ DatanodeDetails target = selectOtherNode(cluster, source);
+ long containerId = createClosedContainer(clientFactory, source, 0L);
+
+ ReplicateContainerCommand cmd =
+ ReplicateContainerCommand.toTarget(containerId, target);
+ queuePushAndWait(cluster, cmd, source,
ReplicationSupervisor::getReplicationSuccessCount);
+ assertNotNull(getContainer(cluster, target, containerId));
+
+ HddsDatanodeService sourceDn = cluster.getHddsDatanode(source);
+ assertVolumePools(sourceDn, DATA_VOLUMES, 1);
+
+ OzoneBucket bucket = TestDataUtil.createVolumeAndBucket(client, VOLUME,
BUCKET);
+ TestDataUtil.createKey(bucket, "key1", RATIS_THREE,
"data".getBytes(UTF_8));
+ OzoneKeyDetails keyDetails = bucket.getKey("key1");
+ 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);
+ }
+ }
+
+ @Test
+ void testHealthyVolumeReplicationAfterVolumeFailure() throws Exception {
+ OzoneConfiguration conf = createPerVolumeConfig(true, 1);
+ configureVolumeFailureDetection(conf);
+ try (MiniOzoneCluster cluster = newCluster(conf, 3);
+ XceiverClientFactory clientFactory = new XceiverClientManager(conf)) {
+ cluster.waitForClusterToBeReady();
+
+ HddsDatanodeService sourceDn = cluster.getHddsDatanodes().get(0);
+ DatanodeDetails source = sourceDn.getDatanodeDetails();
+ DatanodeDetails target = selectOtherNode(cluster, 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);
+
+ VolumeReplicationThreadPools pools =
+
sourceDn.getDatanodeStateMachine().getSupervisor().getVolumeReplicationThreadPools();
+ assertNotNull(pools);
+ assertFalse(pools.hasPool(vol0.getStorageDir().getPath()));
+ assertTrue(pools.hasPool(vol1.getStorageDir().getPath()));
Review Comment:
This assertion can race the async failed-volume listener: MutableVolumeSet
marks the volume failed before running the listener that shuts down per-volume
pools. Waiting for the pool removal (and only then asserting) should reduce
test flakiness.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
