You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ozone.apache.org by pe...@apache.org on 2026/10/22 08:25:44 UTC
(ozone) branch master updated: HDDS-16629. Allow same-rack placement when no other rack has a writable node (#11404)
This is an automated email from the ASF dual-hosted git repository.
peterxcli pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ozone.git
The following commit(s) were added to refs/heads/master by this push:
new 6d11a928722 HDDS-16629. Allow same-rack placement when no other rack has a writable node (#11404)
6d11a928722 is described below
commit 6d11a92872222848f4a7d123e7803745aa00b180
Author: Peter Lee <pe...@gmail.com>
AuthorDate: Thu Oct 22 16:25:38 2026 +0800
HDDS-16629. Allow same-rack placement when no other rack has a writable node (#11404)
Co-authored-by: Siyao Meng <50...@users.noreply.github.com>
---
.../algorithms/SCMContainerPlacementRackAware.java | 20 +++---
.../replication/MisReplicationHandler.java | 23 ++++++-
.../TestSCMContainerPlacementRackAware.java | 44 ++++++++++++
.../replication/MisReplicationHandlerTests.java | 14 +++-
.../TestRatisMisReplicationHandler.java | 80 ++++++++++++++++++++++
5 files changed, 167 insertions(+), 14 deletions(-)
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/SCMContainerPlacementRackAware.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/SCMContainerPlacementRackAware.java
index 9f3bd557fa7..12f335eb1c4 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/SCMContainerPlacementRackAware.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/SCMContainerPlacementRackAware.java
@@ -465,6 +465,7 @@ private DatanodeDetails chooseNode(List<DatanodeDetails> excludedNodes,
steps.add(new Step(null, 0, true));
}
+ boolean outOfRetries = false;
for (Step step : steps) {
int retries = MAX_RETRY;
while (retries > 0) {
@@ -490,15 +491,16 @@ private DatanodeDetails chooseNode(List<DatanodeDetails> excludedNodes,
}
retries--;
}
- // Decommissioned and maintenance nodes stay in the topology, so a rack can use up all its tries on them.
- // If that happens on an affinity node's rack, move on to the next place. Anywhere else, give up.
- if (retries == 0 && step.affinityNode == null) {
- String errMsg = "No writable datanode with enough space was found. "
- + "metadata size required: " + metadataSizeRequired +
- " data size required: " + dataSizeRequired;
- LOG.info(errMsg);
- throw new SCMException(errMsg, null);
- }
+ // Decommissioned and maintenance nodes stay in the topology, so a place can use up all its tries on them,
+ // e.g. the only other rack while a whole rack is being decommissioned. Then move on to the next place.
+ outOfRetries |= retries == 0;
+ }
+ if (outOfRetries) {
+ String errMsg = "No writable datanode with enough space was found. "
+ + "metadata size required: " + metadataSizeRequired +
+ " data size required: " + dataSizeRequired;
+ LOG.info(errMsg);
+ throw new SCMException(errMsg, null);
}
throw new SCMException("No satisfied datanode to meet the excludedNodes and affinityNode constrains.", null);
}
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/MisReplicationHandler.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/MisReplicationHandler.java
index c4be97384e8..a9834ec25d3 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/MisReplicationHandler.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/MisReplicationHandler.java
@@ -29,6 +29,7 @@
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos;
+import org.apache.hadoop.hdds.scm.ContainerPlacementStatus;
import org.apache.hadoop.hdds.scm.PlacementPolicy;
import org.apache.hadoop.hdds.scm.ScmConfigKeys;
import org.apache.hadoop.hdds.scm.container.ContainerInfo;
@@ -133,8 +134,9 @@ public int processAndSendCommands(
List<DatanodeDetails> usedDns = replicas.stream()
.map(ContainerReplica::getDatanodeDetails)
.collect(Collectors.toList());
- if (containerPlacement.validateContainerPlacement(usedDns,
- usedDns.size()).isPolicySatisfied()) {
+ ContainerPlacementStatus placementStatus =
+ containerPlacement.validateContainerPlacement(usedDns, usedDns.size());
+ if (placementStatus.isPolicySatisfied()) {
LOG.info("Container {} is currently not misreplicated",
container.getContainerID());
return 0;
@@ -158,6 +160,23 @@ public int processAndSendCommands(
excludedAndUsedNodes.getUsedNodes(),
excludedAndUsedNodes.getExcludedNodes(), currentContainerSize,
container);
+ if (!targetDatanodes.isEmpty()) {
+ // When no other rack has a usable node, placement can fall back to a rack that already has a replica.
+ // Copying there doesn't help: the container is still mis-replicated, the extra copy gets deleted, and the
+ // next run copies it again, over and over.
+ List<DatanodeDetails> dnsAfterCopy = new ArrayList<>(usedDns);
+ replicasToBeReplicated.stream().limit(targetDatanodes.size())
+ .forEach(r -> dnsAfterCopy.remove(r.getDatanodeDetails()));
+ dnsAfterCopy.addAll(targetDatanodes);
+ ContainerPlacementStatus placementAfterCopy =
+ containerPlacement.validateContainerPlacement(dnsAfterCopy, usedDns.size());
+ if (!placementAfterCopy.isPolicySatisfied()
+ && placementAfterCopy.misReplicationCount() >= placementStatus.misReplicationCount()) {
+ LOG.debug("Not copying container {} to {}, as that would not reduce its mis-replication.",
+ container.getContainerID(), targetDatanodes);
+ return 0;
+ }
+ }
List<DatanodeDetails> availableSources = sources.stream()
.map(ContainerReplica::getDatanodeDetails)
.collect(Collectors.toList());
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/TestSCMContainerPlacementRackAware.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/TestSCMContainerPlacementRackAware.java
index 264d7e61d46..d743a204c9e 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/TestSCMContainerPlacementRackAware.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/TestSCMContainerPlacementRackAware.java
@@ -738,6 +738,50 @@ public void fallbackDoesNotSkipRackOfExcludedNode() throws SCMException {
assertNotEquals(datanodes.get(10), datanodeDetails.get(0));
}
+ @Test
+ public void fallbackToSameRackWhenOtherRackHasOnlyDecommissionedNodes() throws SCMException {
+ setup(2 * NODE_PER_RACK);
+ // All of rack1 is being decommissioned. node5 has a replica that is being decommissioned, so it is excluded,
+ // and nodes 6-9 are already decommissioned.
+ dnInfos.get(5).setNodeStatus(NodeStatus.valueOf(DECOMMISSIONING, HEALTHY));
+ for (int i = 6; i < 10; i++) {
+ dnInfos.get(i).setNodeStatus(NodeStatus.valueOf(DECOMMISSIONED, HEALTHY));
+ }
+ // The other two replicas are on rack0.
+ List<DatanodeDetails> usedNodes = new ArrayList<>(Arrays.asList(datanodes.get(0), datanodes.get(1)));
+ List<DatanodeDetails> excludedNodes = new ArrayList<>(Arrays.asList(datanodes.get(5)));
+
+ List<DatanodeDetails> datanodeDetails = policy.chooseDatanodes(usedNodes, excludedNodes, null, 1, 0, 5);
+
+ assertEquals(1, datanodeDetails.size());
+ // rack1 has no usable node, so the new replica has to go to rack0 as well.
+ assertTrue(cluster.isSameParent(datanodes.get(0), datanodeDetails.get(0)));
+ assertThat(usedNodes).doesNotContain(datanodeDetails.get(0));
+ assertEquals(1, metrics.getDatanodeChooseFallbackCount());
+ // Without fallback, there is nowhere to put it.
+ assertThrows(SCMException.class, () -> policyNoFallback.chooseDatanodes(usedNodes, excludedNodes, null, 1, 0, 5));
+ }
+
+ @Test
+ public void fallbackThrowsWhenNoRackHasUsableNode() {
+ setup(2 * NODE_PER_RACK);
+ // Two replicas are on rack0, and every other node is decommissioned.
+ for (int i = 2; i < 2 * NODE_PER_RACK; i++) {
+ dnInfos.get(i).setNodeStatus(NodeStatus.valueOf(DECOMMISSIONED, HEALTHY));
+ }
+ List<DatanodeDetails> usedNodes = new ArrayList<>(Arrays.asList(datanodes.get(0), datanodes.get(1)));
+
+ // The other-racks step and then the any-rack step use up their 3 tries each.
+ SCMException e = assertThrows(SCMException.class, () -> policy.chooseDatanodes(usedNodes, null, null, 1, 0, 5));
+ assertThat(e.getMessage()).startsWith("No writable datanode");
+ assertEquals(6, metrics.getDatanodeChooseAttemptCount());
+
+ // The same when the any-rack step runs out of nodes to try before it uses up its tries.
+ List<DatanodeDetails> excludedNodes = new ArrayList<>(datanodes.subList(2, NODE_PER_RACK));
+ e = assertThrows(SCMException.class, () -> policy.chooseDatanodes(usedNodes, excludedNodes, null, 1, 0, 5));
+ assertThat(e.getMessage()).startsWith("No writable datanode");
+ }
+
@Test
public void chooseSingleNodeRackWithUsedAndExcludeNodes()
throws SCMException {
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/MisReplicationHandlerTests.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/MisReplicationHandlerTests.java
index 4d3c42d84f4..1a8e21c999f 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/MisReplicationHandlerTests.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/MisReplicationHandlerTests.java
@@ -69,6 +69,8 @@
*/
public abstract class MisReplicationHandlerTests {
+ private static final String TARGET_RACK = "/target-rack";
+
private ContainerInfo container;
private OzoneConfiguration conf;
private ReplicationManager replicationManager;
@@ -125,8 +127,14 @@ static PlacementPolicy mockPlacementPolicy() {
PlacementPolicy placementPolicy = mock(PlacementPolicy.class);
ContainerPlacementStatus mockedContainerPlacementStatus = mock(ContainerPlacementStatus.class);
when(mockedContainerPlacementStatus.isPolicySatisfied()).thenReturn(false);
- when(placementPolicy.validateContainerPlacement(anyList(),
- anyInt())).thenReturn(mockedContainerPlacementStatus);
+ // testMisReplication puts the targets on a rack of their own, and moving a replica there fixes the placement.
+ ContainerPlacementStatus placementAfterCopy = mock(ContainerPlacementStatus.class);
+ when(placementAfterCopy.isPolicySatisfied()).thenReturn(true);
+ when(placementPolicy.validateContainerPlacement(anyList(), anyInt())).thenAnswer(invocation -> {
+ List<DatanodeDetails> dns = invocation.getArgument(0);
+ return dns.stream().anyMatch(dn -> TARGET_RACK.equals(dn.getNetworkLocation()))
+ ? placementAfterCopy : mockedContainerPlacementStatus;
+ });
return placementPolicy;
}
@@ -199,7 +207,7 @@ protected void testMisReplication(Set<ContainerReplica> availableReplicas,
.collect(Collectors.toSet());
List<DatanodeDetails> targetNodes =
IntStream.range(0, expectedNumberOfNodes)
- .mapToObj(i -> MockDatanodeDetails.randomDatanodeDetails())
+ .mapToObj(i -> MockDatanodeDetails.createDatanodeDetails("target" + i, TARGET_RACK))
.collect(Collectors.toList());
if (expectedNumberOfNodes > 0) {
when(mockedPlacementPolicy.chooseDatanodes(
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestRatisMisReplicationHandler.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestRatisMisReplicationHandler.java
index e5b25f8be4d..47352876009 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestRatisMisReplicationHandler.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestRatisMisReplicationHandler.java
@@ -17,8 +17,10 @@
package org.apache.hadoop.hdds.scm.container.replication;
+import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeOperationalState.DECOMMISSIONED;
import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeOperationalState.IN_MAINTENANCE;
import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeOperationalState.IN_SERVICE;
+import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeState.HEALTHY;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.Mockito.any;
@@ -26,12 +28,16 @@
import static org.mockito.Mockito.anyList;
import static org.mockito.Mockito.anyLong;
import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.eq;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.io.File;
import java.io.IOException;
+import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -40,19 +46,32 @@
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.protocol.MockDatanodeDetails;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos.LifeCycleState;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor;
import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerReplicaProto.State;
import org.apache.hadoop.hdds.scm.ContainerPlacementStatus;
+import org.apache.hadoop.hdds.scm.HddsTestUtils;
import org.apache.hadoop.hdds.scm.PlacementPolicy;
+import org.apache.hadoop.hdds.scm.container.ContainerInfo;
import org.apache.hadoop.hdds.scm.container.ContainerReplica;
+import org.apache.hadoop.hdds.scm.container.placement.algorithms.SCMContainerPlacementMetrics;
+import org.apache.hadoop.hdds.scm.container.placement.algorithms.SCMContainerPlacementRackAware;
import org.apache.hadoop.hdds.scm.exceptions.SCMException;
+import org.apache.hadoop.hdds.scm.net.NetworkTopology;
+import org.apache.hadoop.hdds.scm.net.NetworkTopologyImpl;
+import org.apache.hadoop.hdds.scm.net.NodeSchemaManager;
+import org.apache.hadoop.hdds.scm.node.DatanodeInfo;
+import org.apache.hadoop.hdds.scm.node.NodeManager;
+import org.apache.hadoop.hdds.scm.node.NodeStatus;
import org.apache.hadoop.hdds.scm.node.states.NodeNotFoundException;
+import org.apache.hadoop.ozone.container.upgrade.UpgradeUtils;
import org.apache.ratis.protocol.exceptions.NotLeaderException;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.ArgumentCaptor;
/**
* Tests the RatisReplicationHandling functionality.
@@ -173,6 +192,67 @@ public void testMisReplicationWithPendingOps()
pendingOp, 0, 1, 0);
}
+ @Test
+ public void testCopyIsSkippedUntilItReducesMisReplication() throws IOException {
+ Set<ContainerReplica> availableReplicas = ReplicationTestUtil
+ .createReplicas(Pair.of(IN_SERVICE, 0), Pair.of(IN_SERVICE, 0), Pair.of(IN_SERVICE, 0));
+ // The target doesn't help, e.g. because it is on the same rack as all the existing replicas.
+ PlacementPolicy placementPolicy = mock(PlacementPolicy.class);
+ ContainerPlacementStatus mockedContainerPlacementStatus = mock(ContainerPlacementStatus.class);
+ when(mockedContainerPlacementStatus.isPolicySatisfied()).thenReturn(false);
+ when(mockedContainerPlacementStatus.misReplicationCount()).thenReturn(1);
+ when(placementPolicy.validateContainerPlacement(anyList(), anyInt())).thenReturn(mockedContainerPlacementStatus);
+ testMisReplication(availableReplicas, placementPolicy, Collections.emptyList(), 0, 1, 1, 0);
+
+ // On a later run, placement finds a target on another rack, so the copy is sent.
+ testMisReplication(availableReplicas, mockPlacementPolicy(), Collections.emptyList(), 0, 1, 1);
+ }
+
+ @Test
+ public void testCopyWaitsForUsableNodeOnAnotherRack() throws IOException {
+ // rack0 has the replicas on node0-2 and a free node3. Both nodes of rack1 are decommissioned.
+ NetworkTopology topology = new NetworkTopologyImpl(NodeSchemaManager.getInstance());
+ NodeManager nodeManager = mock(NodeManager.class);
+ when(nodeManager.getClusterNetworkTopologyMap()).thenReturn(topology);
+ when(nodeManager.hasAvailableSpace(any())).thenReturn(true);
+ List<DatanodeDetails> nodes = new ArrayList<>();
+ List<DatanodeInfo> nodeInfos = new ArrayList<>();
+ for (int i = 0; i < 6; i++) {
+ DatanodeDetails node = MockDatanodeDetails.createDatanodeDetails("node" + i, "/rack" + i / 4);
+ topology.add(node);
+ NodeStatus status = i < 4 ? NodeStatus.inServiceHealthy() : NodeStatus.valueOf(DECOMMISSIONED, HEALTHY);
+ DatanodeInfo nodeInfo =
+ new DatanodeInfo(node, status, UpgradeUtils.defaultVersionProto(), HddsTestUtils.ROLL_INTERVAL_MS_DEFAULT);
+ when(nodeManager.getNode(node.getID())).thenReturn(nodeInfo);
+ nodes.add(node);
+ nodeInfos.add(nodeInfo);
+ }
+ PlacementPolicy placementPolicy = new SCMContainerPlacementRackAware(nodeManager, new OzoneConfiguration(),
+ topology, true, mock(SCMContainerPlacementMetrics.class));
+ ContainerInfo container = ReplicationTestUtil.createContainer(LifeCycleState.CLOSED,
+ RatisReplicationConfig.getInstance(ReplicationFactor.THREE));
+ Set<ContainerReplica> replicas = new HashSet<>();
+ for (DatanodeDetails node : nodes.subList(0, 3)) {
+ replicas.add(ReplicationTestUtil.createContainerReplica(container.containerID(), 0, IN_SERVICE, State.CLOSED,
+ 123L, 1234L, node, node.getID()));
+ }
+ ContainerHealthResult.MisReplicatedHealthResult result =
+ mock(ContainerHealthResult.MisReplicatedHealthResult.class);
+ when(result.getContainerInfo()).thenReturn(container);
+ MisReplicationHandler handler =
+ getMisreplicationHandler(placementPolicy, new OzoneConfiguration(), getReplicationManager());
+
+ // Placement can only offer node3, on the same rack as the replicas.
+ assertEquals(0, handler.processAndSendCommands(replicas, Collections.emptyList(), result, 0));
+
+ // node4 on rack1 is back in service.
+ nodeInfos.get(4).setNodeStatus(NodeStatus.inServiceHealthy());
+ assertEquals(1, handler.processAndSendCommands(replicas, Collections.emptyList(), result, 0));
+ ArgumentCaptor<DatanodeDetails> target = ArgumentCaptor.forClass(DatanodeDetails.class);
+ verify(getReplicationManager()).sendThrottledReplicationCommand(eq(container), anyList(), target.capture(), eq(0));
+ assertEquals(nodes.get(4), target.getValue());
+ }
+
@Test
public void testAllSourcesOverloaded() throws IOException {
ReplicationManager replicationManager = getReplicationManager();
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@ozone.apache.org
For additional commands, e-mail: commits-help@ozone.apache.org