diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/RatisPipelineProvider.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/RatisPipelineProvider.java index 6563e3fcaa..df21420be1 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/RatisPipelineProvider.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/RatisPipelineProvider.java @@ -91,7 +91,9 @@ private static ContainerPlacementPolicy createContainerPlacementPolicy( public Pipeline create(ReplicationFactor factor) throws IOException { // Get set of datanodes already used for ratis pipeline Set dnsUsed = new HashSet<>(); - stateManager.getPipelines(ReplicationType.RATIS, factor) + stateManager.getPipelines(ReplicationType.RATIS, factor).stream().filter( + p -> p.getPipelineState().equals(PipelineState.OPEN) || + p.getPipelineState().equals(PipelineState.ALLOCATED)) .forEach(p -> dnsUsed.addAll(p.getNodes())); // Get list of healthy nodes diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/MockNodeManager.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/MockNodeManager.java index 129644e7de..c10bc44068 100644 --- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/MockNodeManager.java +++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/MockNodeManager.java @@ -43,6 +43,7 @@ import org.assertj.core.util.Preconditions; import java.io.IOException; +import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; import java.util.LinkedList; @@ -184,7 +185,7 @@ public int getNodeCount(HddsProtos.NodeState nodestate) { */ @Override public List getAllNodes() { - return null; + return new ArrayList<>(nodeMetricMap.keySet()); } /** diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestRatisPipelineProvider.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestRatisPipelineProvider.java index 28f47cc80f..00144e4e65 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestRatisPipelineProvider.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestRatisPipelineProvider.java @@ -135,4 +135,67 @@ public void testCreatePipelineWithNodes() { Pipeline.PipelineState.OPEN); Assert.assertEquals(pipeline.getNodes().size(), factor.getNumber()); } + + @Test + public void testCreatePipelinesDnExclude() throws IOException { + + // We have 10 DNs in MockNodeManager. + // Use up first 3 DNs for an open pipeline. + List openPiplineDns = nodeManager.getAllNodes() + .subList(0, 3); + HddsProtos.ReplicationFactor factor = HddsProtos.ReplicationFactor.THREE; + + Pipeline openPipeline = Pipeline.newBuilder() + .setType(HddsProtos.ReplicationType.RATIS) + .setFactor(factor) + .setNodes(openPiplineDns) + .setState(Pipeline.PipelineState.OPEN) + .setId(PipelineID.randomId()) + .build(); + + stateManager.addPipeline(openPipeline); + + // Use up next 3 DNs also for an open pipeline. + List moreOpenPiplineDns = nodeManager.getAllNodes() + .subList(3, 6); + Pipeline anotherOpenPipeline = Pipeline.newBuilder() + .setType(HddsProtos.ReplicationType.RATIS) + .setFactor(factor) + .setNodes(moreOpenPiplineDns) + .setState(Pipeline.PipelineState.OPEN) + .setId(PipelineID.randomId()) + .build(); + stateManager.addPipeline(anotherOpenPipeline); + + // Use up next 3 DNs also for a closed pipeline. + List closedPiplineDns = nodeManager.getAllNodes() + .subList(6, 9); + Pipeline anotherClosedPipeline = Pipeline.newBuilder() + .setType(HddsProtos.ReplicationType.RATIS) + .setFactor(factor) + .setNodes(closedPiplineDns) + .setState(Pipeline.PipelineState.CLOSED) + .setId(PipelineID.randomId()) + .build(); + stateManager.addPipeline(anotherClosedPipeline); + + Pipeline pipeline = provider.create(factor); + Assert.assertEquals(pipeline.getType(), HddsProtos.ReplicationType.RATIS); + Assert.assertEquals(pipeline.getFactor(), factor); + Assert.assertEquals(pipeline.getPipelineState(), + Pipeline.PipelineState.OPEN); + Assert.assertEquals(pipeline.getNodes().size(), factor.getNumber()); + List pipelineNodes = pipeline.getNodes(); + + // Pipline nodes cannot be from open pipelines. + Assert.assertTrue( + pipelineNodes.parallelStream().filter(dn -> + (openPiplineDns.contains(dn) || moreOpenPiplineDns.contains(dn))) + .count() == 0); + + // Since we have only 10 DNs, at least 1 pipeline node should have been + // from the closed pipeline DN list. + Assert.assertTrue(pipelineNodes.parallelStream().filter( + closedPiplineDns::contains).count() > 0); + } } \ No newline at end of file