HDFS-15363. BlockPlacementPolicyWithNodeGroup should validate if it is initialized by NetworkTopologyWithNodeGroup. Contributed by hemanthboyina.

(cherry picked from commit 4d22d1c58f)

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockPlacementPolicyWithNodeGroup.java

@@ -46,6 +46,11 @@ public class BlockPlacementPolicyWithNodeGroup extends BlockPlacementPolicyDefau
   public void initialize(Configuration conf,  FSClusterStats stats,
           NetworkTopology clusterMap, 
           Host2NodesMap host2datanodeMap) {
+    if (!(clusterMap instanceof NetworkTopologyWithNodeGroup)) {
+      throw new IllegalArgumentException(
+          "Configured cluster topology should be "
+              + NetworkTopologyWithNodeGroup.class.getName());
+    }
     super.initialize(conf, stats, clusterMap, host2datanodeMap);
   }
 

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/balancer/TestBalancerWithNodeGroup.java

@@ -47,6 +47,7 @@ import org.apache.hadoop.hdfs.protocol.HdfsConstants.DatanodeReportType;
 import org.apache.hadoop.hdfs.server.blockmanagement.BlockPlacementPolicyWithNodeGroup;
 import org.apache.hadoop.net.NetworkTopology;
 import org.apache.hadoop.net.NetworkTopologyWithNodeGroup;
+import org.apache.hadoop.test.LambdaTestUtils;
 import org.junit.Assert;
 import org.junit.Test;
 
@@ -361,4 +362,30 @@ public class TestBalancerWithNodeGroup {
       cluster.shutdown();
     }
   }
+
+  /**
+   * verify BlockPlacementPolicyNodeGroup uses NetworkTopologyWithNodeGroup.
+   */
+
+  @Test
+  public void testBPPNodeGroup() throws Exception {
+    Configuration conf = createConf();
+    conf.setBoolean(DFSConfigKeys.DFS_USE_DFS_NETWORK_TOPOLOGY_KEY, true);
+    long[] capacities = new long[] {CAPACITY, CAPACITY, CAPACITY, CAPACITY};
+    String[] racks = new String[] {RACK0, RACK0, RACK1, RACK1};
+    String[] nodeGroups =
+        new String[] {NODEGROUP0, NODEGROUP0, NODEGROUP1, NODEGROUP2};
+
+    int numOfDatanodes = capacities.length;
+    assertEquals(numOfDatanodes, racks.length);
+    assertEquals(numOfDatanodes, nodeGroups.length);
+    MiniDFSCluster.Builder builder =
+        new MiniDFSCluster.Builder(conf).numDataNodes(capacities.length)
+            .racks(racks).simulatedCapacities(capacities);
+    MiniDFSClusterWithNodeGroup.setNodeGroups(nodeGroups);
+    LambdaTestUtils.intercept(IllegalArgumentException.class,
+        "Configured cluster topology should be "
+            + "org.apache.hadoop.net.NetworkTopologyWithNodeGroup",
+        () -> new MiniDFSClusterWithNodeGroup(builder));
+  }
 }