HDDS-8383. Misreplication cannot be resolved with single rack

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/net/NetworkTopology.java

@@ -242,4 +242,10 @@ Node getNode(int leafIndex, String scope, List<String> excludedScopes,
    */
   List<? extends Node> sortByDistanceCost(Node reader,
       List<? extends Node> nodes, int activeLen);
+
+  default int getRackCount() {
+    // The leaf nodes are all at max level, so the number of nodes at
+    // leafLevel - 1 is the rack count
+    return getNumOfNodes(getMaxLevel() - 1);
+  }
 }

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/SCMCommonPlacementPolicy.java

@@ -74,9 +74,9 @@ public abstract class SCMCommonPlacementPolicy implements
    * or the replication factor is 1 or the required racks is 1.
    */
   private ContainerPlacementStatus validPlacement
-      = new ContainerPlacementStatusDefault(1, 1, 1);
+      = new ContainerPlacementStatusDefault(1, 1);
   private ContainerPlacementStatus invalidPlacement
-      = new ContainerPlacementStatusDefault(0, 1, 1);
+      = new ContainerPlacementStatusDefault(0, 1);
 
   /**
    * Constructor.
@@ -402,17 +402,12 @@ public ContainerPlacementStatus validateContainerPlacement(
     Map<Node, Long> currentRackCount = dns.stream()
             .collect(Collectors.groupingBy(this::getPlacementGroup,
                     Collectors.counting()));
-    final int maxLevel = topology.getMaxLevel();
-    // The leaf nodes are all at max level, so the number of nodes at
-    // leafLevel - 1 is the rack count
-    int numRacks = topology.getNumOfNodes(maxLevel - 1);
     if (replicas < requiredRacks) {
       requiredRacks = replicas;
     }
-    int maxReplicasPerRack = getMaxReplicasPerRack(replicas,
-            Math.min(requiredRacks, numRacks));
+    int maxReplicasPerRack = getMaxReplicasPerRack(replicas, requiredRacks);
     return new ContainerPlacementStatusDefault(
-        currentRackCount.size(), requiredRacks, numRacks, maxReplicasPerRack,
+        currentRackCount.size(), requiredRacks, maxReplicasPerRack,
             currentRackCount.values().stream().map(Long::intValue)
                     .collect(Collectors.toList()));
   }

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/ContainerPlacementStatusDefault.java

@@ -30,31 +30,28 @@ public class ContainerPlacementStatusDefault
 
   private final int requiredRacks;
   private final int currentRacks;
-  private final int totalRacks;
 
   private final int maxReplicasPerRack;
   private final List<Integer> rackReplicaCnts;
 
 
   public ContainerPlacementStatusDefault(int currentRacks, int requiredRacks,
-      int totalRacks, int maxReplicasPerRack, List<Integer> rackReplicaCnts) {
+      int maxReplicasPerRack, List<Integer> rackReplicaCnts) {
     this.requiredRacks = requiredRacks;
     this.currentRacks = currentRacks;
-    this.totalRacks = totalRacks;
     this.maxReplicasPerRack = maxReplicasPerRack;
     this.rackReplicaCnts = rackReplicaCnts;
   }
 
-  public ContainerPlacementStatusDefault(int currentRacks, int requiredRacks,
-      int totalRacks) {
-    this(currentRacks, requiredRacks, totalRacks, 1,
+  public ContainerPlacementStatusDefault(int currentRacks, int requiredRacks) {
+    this(currentRacks, requiredRacks, 1,
          currentRacks == 0 ? Collections.emptyList()
                  : Collections.nCopies(currentRacks, 1));
   }
 
   @Override
   public boolean isPolicySatisfied() {
-    if (currentRacks < Math.min(totalRacks, requiredRacks)) {
+    if (currentRacks < requiredRacks) {
       return false;
     }
     return rackReplicaCnts.stream().allMatch(cnt -> cnt <= maxReplicasPerRack);
@@ -86,7 +83,7 @@ public int misReplicationCount() {
 
   @Override
   public int expectedPlacementCount() {
-    return Math.min(requiredRacks, totalRacks);
+    return requiredRacks;
   }
 
   @Override

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/SCMContainerPlacementRackAware.java

@@ -386,6 +386,7 @@ protected int getMaxReplicasPerRack(int numReplicas, int numberOfRacks) {
 
   @Override
   protected int getRequiredRackCount(int numReplicas) {
-    return REQUIRED_RACKS;
+    int racks = networkTopology != null ? networkTopology.getRackCount() : 1;
+    return Math.min(REQUIRED_RACKS, racks);
   }
 }

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/SCMContainerPlacementRackScatter.java

@@ -438,8 +438,7 @@ protected int getRequiredRackCount(int numReplicas) {
     if (networkTopology == null) {
       return 1;
     }
-    int maxLevel = networkTopology.getMaxLevel();
-    int numRacks = networkTopology.getNumOfNodes(maxLevel - 1);
+    int numRacks = networkTopology.getRackCount();
     // Return the num of Rack if numRack less than numReplicas
     return Math.min(numRacks, numReplicas);
   }

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelinePlacementPolicy.java

@@ -475,15 +475,14 @@ protected DatanodeDetails chooseNodeBasedOnSameRack(
    * @return true when all nodes are equal
    */
   private boolean checkAllNodesAreEqual(NetworkTopology topology) {
-    if (topology == null) {
-      return true;
-    }
-    return (topology.getNumOfNodes(topology.getMaxLevel() - 1) == 1);
+    return topology == null || topology.getRackCount() == 1;
   }
 
   @Override
   protected int getRequiredRackCount(int numReplicas) {
-    return REQUIRED_RACKS;
+    NetworkTopology topology = nodeManager.getClusterNetworkTopologyMap();
+    int racks = topology != null ? topology.getRackCount() : 1;
+    return Math.min(REQUIRED_RACKS, racks);
   }
 
   /**

hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/TestContainerPlacementFactory.java

@@ -194,7 +194,7 @@ public List<DatanodeDetails> chooseDatanodes(
     @Override
     public ContainerPlacementStatus
         validateContainerPlacement(List<DatanodeDetails> dns, int replicas) {
-      return new ContainerPlacementStatusDefault(1, 1, 1);
+      return new ContainerPlacementStatusDefault(1, 1);
     }
 
     @Override

hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/TestContainerPlacementStatusDefault.java

@@ -35,53 +35,44 @@ public class TestContainerPlacementStatusDefault {
   @Test
   public void testPlacementSatisfiedCorrectly() {
     ContainerPlacementStatusDefault stat =
-        new ContainerPlacementStatusDefault(1, 1, 1);
+        new ContainerPlacementStatusDefault(1, 1);
     assertTrue(stat.isPolicySatisfied());
     assertEquals(0, stat.misReplicationCount());
 
-    // Requires 2 racks, but cluster only has 1
-    stat = new ContainerPlacementStatusDefault(1, 2, 1);
+    stat = new ContainerPlacementStatusDefault(2, 2);
     assertTrue(stat.isPolicySatisfied());
     assertEquals(0, stat.misReplicationCount());
 
-    stat = new ContainerPlacementStatusDefault(2, 2, 3);
-    assertTrue(stat.isPolicySatisfied());
-    assertEquals(0, stat.misReplicationCount());
-
-    stat = new ContainerPlacementStatusDefault(3, 2, 3);
-    assertTrue(stat.isPolicySatisfied());
-    assertEquals(0, stat.misReplicationCount());
-
-    stat = new ContainerPlacementStatusDefault(3, 2, 3);
+    stat = new ContainerPlacementStatusDefault(3, 2);
     assertTrue(stat.isPolicySatisfied());
     assertEquals(0, stat.misReplicationCount());
   }
 
   @Test
   public void testPlacementNotSatisfied() {
     ContainerPlacementStatusDefault stat =
-        new ContainerPlacementStatusDefault(1, 2, 2);
+        new ContainerPlacementStatusDefault(1, 2);
     assertFalse(stat.isPolicySatisfied());
     assertEquals(1, stat.misReplicationCount());
 
     // Zero rack, but need 2 - shouldn't really happen in practice
-    stat = new ContainerPlacementStatusDefault(0, 2, 1);
+    stat = new ContainerPlacementStatusDefault(0, 2);
     assertFalse(stat.isPolicySatisfied());
-    assertEquals(1, stat.misReplicationCount());
+    assertEquals(2, stat.misReplicationCount());
 
-    stat = new ContainerPlacementStatusDefault(2, 3, 3);
+    stat = new ContainerPlacementStatusDefault(2, 3);
     assertFalse(stat.isPolicySatisfied());
     assertEquals(1, stat.misReplicationCount());
 
-    stat = new ContainerPlacementStatusDefault(2, 4, 3, 1, Arrays.asList(1, 3));
+    stat = new ContainerPlacementStatusDefault(2, 4, 1, Arrays.asList(1, 3));
     assertFalse(stat.isPolicySatisfied());
     assertEquals(2, stat.misReplicationCount());
 
-    stat = new ContainerPlacementStatusDefault(1, 4, 3, 1, Arrays.asList(1, 2));
+    stat = new ContainerPlacementStatusDefault(1, 4, 1, Arrays.asList(1, 2));
     assertFalse(stat.isPolicySatisfied());
-    assertEquals(2, stat.misReplicationCount());
+    assertEquals(3, stat.misReplicationCount());
 
-    stat = new ContainerPlacementStatusDefault(2, 2, 3, 2, Arrays.asList(3, 1));
+    stat = new ContainerPlacementStatusDefault(2, 2, 2, Arrays.asList(3, 1));
     assertFalse(stat.isPolicySatisfied());
     assertEquals(1, stat.misReplicationCount());
 

hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestECUnderReplicationHandler.java

@@ -135,7 +135,7 @@ public NodeStatus getNodeStatus(DatanodeDetails dd) {
     ecPlacementPolicy = Mockito.mock(PlacementPolicy.class);
     Mockito.when(ecPlacementPolicy.validateContainerPlacement(
         anyList(), anyInt()))
-        .thenReturn(new ContainerPlacementStatusDefault(2, 2, 3));
+        .thenReturn(new ContainerPlacementStatusDefault(2, 2));
   }
 
   @Test

hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestLegacyReplicationManager.java

@@ -220,7 +220,7 @@ public void setup()
         Mockito.any(),
         Mockito.anyInt()
         )).thenAnswer(invocation ->
-        new ContainerPlacementStatusDefault(2, 2, 3));
+        new ContainerPlacementStatusDefault(2, 2));
     clock = new TestClock(Instant.now(), ZoneId.of("UTC"));
     containerReplicaPendingOps = new ContainerReplicaPendingOps(conf, clock);
     createReplicationManager(new ReplicationManagerConfiguration());
@@ -2217,7 +2217,7 @@ public void additionalReplicaScheduledWhenMisReplicated()
               Mockito.argThat(list -> list.size() == 3),
               Mockito.anyInt()
       )).thenAnswer(invocation -> {
-        return new ContainerPlacementStatusDefault(1, 2, 3);
+        return new ContainerPlacementStatusDefault(1, 2);
       });
 
       int currentReplicateCommandCount = datanodeCommandHandler
@@ -2253,7 +2253,7 @@ public void additionalReplicaScheduledWhenMisReplicated()
               Mockito.anyList(),
               Mockito.anyInt()
       )).thenAnswer(invocation -> {
-        return new ContainerPlacementStatusDefault(1, 2, 3);
+        return new ContainerPlacementStatusDefault(1, 2);
       });
 
       currentReplicateCommandCount = datanodeCommandHandler.getInvocationCount(
@@ -2306,7 +2306,7 @@ public void overReplicatedButRemovingMakesMisReplicated()
               Mockito.argThat(list -> list.size() == 3),
               Mockito.anyInt()
       )).thenAnswer(
-              invocation -> new ContainerPlacementStatusDefault(1, 2, 3));
+              invocation -> new ContainerPlacementStatusDefault(1, 2));
 
       int currentDeleteCommandCount = datanodeCommandHandler
               .getInvocationCount(SCMCommandProto.Type.deleteContainerCommand);
@@ -2362,7 +2362,7 @@ public void testOverReplicatedAndPolicySatisfied()
               Mockito.argThat(list -> list.size() == 3),
               Mockito.anyInt()
       )).thenAnswer(
-              invocation -> new ContainerPlacementStatusDefault(2, 2, 3));
+              invocation -> new ContainerPlacementStatusDefault(2, 2));
 
       final int currentDeleteCommandCount = datanodeCommandHandler
               .getInvocationCount(SCMCommandProto.Type.deleteContainerCommand);
@@ -2410,7 +2410,7 @@ public void testOverReplicatedAndPolicyUnSatisfiedAndDeleted()
               Mockito.argThat(list -> list != null && list.size() <= 4),
               Mockito.anyInt()
       )).thenAnswer(
-              invocation -> new ContainerPlacementStatusDefault(1, 2, 3));
+              invocation -> new ContainerPlacementStatusDefault(1, 2));
 
       int currentDeleteCommandCount = datanodeCommandHandler
               .getInvocationCount(SCMCommandProto.Type.deleteContainerCommand);

hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestRatisOverReplicationHandler.java

@@ -82,7 +82,7 @@ public void setup() throws NodeNotFoundException, NotLeaderException,
     policy = Mockito.mock(PlacementPolicy.class);
     Mockito.when(policy.validateContainerPlacement(
         Mockito.anyList(), Mockito.anyInt()))
-        .thenReturn(new ContainerPlacementStatusDefault(2, 2, 3));
+        .thenReturn(new ContainerPlacementStatusDefault(2, 2));
 
     replicationManager = Mockito.mock(ReplicationManager.class);
     Mockito.when(replicationManager.getNodeStatus(any(DatanodeDetails.class)))
@@ -195,7 +195,7 @@ public void testOverReplicatedContainerBecomesMisReplicatedOnRemoving()
     // checked.
     Mockito.when(policy.validateContainerPlacement(
         Mockito.argThat(list -> list.size() <= 4), Mockito.anyInt()))
-        .thenReturn(new ContainerPlacementStatusDefault(1, 2, 3));
+        .thenReturn(new ContainerPlacementStatusDefault(1, 2));
 
     testProcessing(replicas, Collections.emptyList(),
         getOverReplicatedHealthResult(), 0);
@@ -225,7 +225,7 @@ public void testOverReplicatedClosedContainerWithQuasiClosedReplica()
     // checked.
     Mockito.when(policy.validateContainerPlacement(
             Mockito.argThat(list -> list.size() <= 4), Mockito.anyInt()))
-        .thenReturn(new ContainerPlacementStatusDefault(1, 2, 3));
+        .thenReturn(new ContainerPlacementStatusDefault(1, 2));
 
     Set<Pair<DatanodeDetails, SCMCommand<?>>> commands = testProcessing(
         replicas, Collections.emptyList(), getOverReplicatedHealthResult(), 2);

hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java

@@ -114,11 +114,11 @@ public void setup() throws IOException {
     containerManager = Mockito.mock(ContainerManager.class);
     ratisPlacementPolicy = Mockito.mock(PlacementPolicy.class);
     Mockito.when(ratisPlacementPolicy.validateContainerPlacement(anyList(),
-        anyInt())).thenReturn(new ContainerPlacementStatusDefault(2, 2, 3));
+        anyInt())).thenReturn(new ContainerPlacementStatusDefault(2, 2));
     ecPlacementPolicy = Mockito.mock(PlacementPolicy.class);
     Mockito.when(ecPlacementPolicy.validateContainerPlacement(
         anyList(), anyInt()))
-        .thenReturn(new ContainerPlacementStatusDefault(2, 2, 3));
+        .thenReturn(new ContainerPlacementStatusDefault(2, 2));
 
     scmContext = Mockito.mock(SCMContext.class);
 
@@ -626,7 +626,7 @@ public void testUnderReplicationQueuePopulated() {
     // replicated take precedence.
     Mockito.when(ecPlacementPolicy.validateContainerPlacement(
             anyList(), anyInt()))
-        .thenReturn(new ContainerPlacementStatusDefault(1, 2, 3));
+        .thenReturn(new ContainerPlacementStatusDefault(1, 2));
 
     ContainerInfo decomContainer = createContainerInfo(repConfig, 1,
         HddsProtos.LifeCycleState.CLOSED);

hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/health/TestECReplicationCheckHandler.java

@@ -79,7 +79,7 @@ public void setup() {
     placementPolicy = Mockito.mock(PlacementPolicy.class);
     Mockito.when(placementPolicy.validateContainerPlacement(
         anyList(), anyInt()))
-        .thenReturn(new ContainerPlacementStatusDefault(2, 2, 3));
+        .thenReturn(new ContainerPlacementStatusDefault(2, 2));
     healthCheck = new ECReplicationCheckHandler(placementPolicy);
     repConfig = new ECReplicationConfig(3, 2);
     repQueue = new ReplicationQueue();
@@ -551,7 +551,7 @@ public void testMisReplicatedContainer() {
         Mockito.any(),
         Mockito.anyInt()
     )).thenAnswer(invocation ->
-        new ContainerPlacementStatusDefault(4, 5, 9));
+        new ContainerPlacementStatusDefault(4, 5));
 
     Set<ContainerReplica> replicas =  createReplicas(container.containerID(),
         Pair.of(IN_SERVICE, 1), Pair.of(IN_SERVICE, 2),
@@ -587,9 +587,9 @@ public void testMisReplicatedContainerFixedByPending() {
       List<DatanodeDetails> dns = invocation.getArgument(0);
       // If the number of DNs is 5 or less make it be mis-replicated
       if (dns.size() <= 5) {
-        return new ContainerPlacementStatusDefault(4, 5, 9);
+        return new ContainerPlacementStatusDefault(4, 5);
       } else {
-        return new ContainerPlacementStatusDefault(5, 5, 9);
+        return new ContainerPlacementStatusDefault(5, 5);
       }
     });
 
@@ -632,7 +632,7 @@ public void testUnderAndMisReplicatedContainer() {
         Mockito.any(),
         Mockito.anyInt()
     )).thenAnswer(invocation ->
-        new ContainerPlacementStatusDefault(4, 5, 9));
+        new ContainerPlacementStatusDefault(4, 5));
 
     Set<ContainerReplica> replicas =  createReplicas(container.containerID(),
         Pair.of(IN_SERVICE, 1), Pair.of(IN_SERVICE, 2),
@@ -667,7 +667,7 @@ public void testOverAndMisReplicatedContainer() {
         Mockito.any(),
         Mockito.anyInt()
     )).thenAnswer(invocation ->
-        new ContainerPlacementStatusDefault(4, 5, 9));
+        new ContainerPlacementStatusDefault(4, 5));
 
     Set<ContainerReplica> replicas =  createReplicas(container.containerID(),
         Pair.of(IN_SERVICE, 1), Pair.of(IN_SERVICE, 2),