Java 类org.apache.hadoop.hdfs.server.namenode.BlockPlacementPolicy 实例源码

/**
 * Tests replication in DFS.
 */
private void runReplication(boolean simulated,
    Class<? extends BlockPlacementPolicy> clazz) throws IOException {
  Configuration conf = new Configuration();
  conf.setClass("dfs.block.replicator.classname", clazz,
      BlockPlacementPolicy.class);
  conf.setBoolean("dfs.replication.considerLoad", false);
  if (simulated) {
    conf.setBoolean(SimulatedFSDataset.CONFIG_PROPERTY_SIMULATED, true);
  }    
  MiniDFSCluster cluster = null;
  FileSystem fileSys = null;
  DFSClient client = null;
  try {
    cluster = new MiniDFSCluster(conf, numDatanodes, racks, null,
        true, true);
    cluster.waitActive();
    cluster.getNameNode().namesystem.refreshNodes(conf);

    client = new DFSClient(cluster.getNameNode().getNameNodeAddress(), conf);
    DatanodeInfo[] info = client.datanodeReport(DatanodeReportType.LIVE);
    assertEquals("Number of Datanodes ", numDatanodes, info.length);
    fileSys = cluster.getFileSystem();

    Path file1 = new Path("/smallblocktest.dat");
    writeFile(fileSys, file1, 3);
    checkFile(fileSys, file1, 3);
    cleanupFile(fileSys, file1);
    writeFile(fileSys, file1, 10);
    checkFile(fileSys, file1, 10);
    cleanupFile(fileSys, file1);
    writeFile(fileSys, file1, 4);
    checkFile(fileSys, file1, 4);
    cleanupFile(fileSys, file1);
    writeFile(fileSys, file1, 1);
    checkFile(fileSys, file1, 1);
    cleanupFile(fileSys, file1);
    writeFile(fileSys, file1, 2);
    checkFile(fileSys, file1, 2);
    cleanupFile(fileSys, file1);
  } finally {
    if(client != null) {
      client.close();
    }
    if(fileSys != null)
      fileSys.close();
    if (cluster != null)
      cluster.shutdown();
  }
}

private void runPendingReplicationRetry(
    Class<? extends BlockPlacementPolicy> clazz) throws IOException {
  pendingReplicationRetryInternal(false, clazz);
}

private void checkReplicationPolicyCompatibility(Configuration conf) {
  if (!(BlockPlacementPolicy.getInstance(conf, null, null, null, null, null)
      instanceof BlockPlacementPolicyDefault)) {
    throw new IllegalArgumentException("Configuration lacks BlockPlacementPolicyDefault");
  }
}

private void checkReplicationPolicyCompatibility(Configuration conf) throws UnsupportedActionException {
  if (BlockPlacementPolicy.getInstance(conf, null, null).getClass() != 
      BlockPlacementPolicyDefault.class) {
    throw new UnsupportedActionException("Balancer without BlockPlacementPolicyDefault");
  }
}

private void checkReplicationPolicyCompatibility(Configuration conf) {
  if (!(BlockPlacementPolicy.getInstance(conf, null, null, null, null, null)
      instanceof BlockPlacementPolicyDefault)) {
    throw new IllegalArgumentException("Configuration lacks BlockPlacementPolicyDefault");
  }
}