DistCh(Configuration conf) {
    super(createJobConf(conf));
  }
@Test
  public void testDistCh() throws Exception {
    final Configuration conf = new Configuration();

    conf.set(CapacitySchedulerConfiguration.PREFIX+CapacitySchedulerConfiguration.ROOT+"."+CapacitySchedulerConfiguration.QUEUES, "default");
    conf.set(CapacitySchedulerConfiguration.PREFIX+CapacitySchedulerConfiguration.ROOT+".default."+CapacitySchedulerConfiguration.CAPACITY, "100");
    final MiniDFSCluster cluster=  new MiniDFSCluster.Builder(conf).numDataNodes(2).format(true).build();
    
    final FileSystem fs = cluster.getFileSystem();
    final FsShell shell = new FsShell(conf);
    
    try {
      final FileTree tree = new FileTree(fs, "testDistCh");
      final FileStatus rootstatus = fs.getFileStatus(tree.rootdir);

      runLsr(shell, tree.root, 0);

      final String[] args = new String[NUN_SUBS];
      final ChPermissionStatus[] newstatus = new ChPermissionStatus[NUN_SUBS];

      
      args[0]="/test/testDistCh/sub0:sub1::";
      newstatus[0] = new ChPermissionStatus(rootstatus, "sub1", "", "");

      args[1]="/test/testDistCh/sub1::sub2:";
      newstatus[1] = new ChPermissionStatus(rootstatus, "", "sub2", "");

      args[2]="/test/testDistCh/sub2:::437";
      newstatus[2] = new ChPermissionStatus(rootstatus, "", "", "437");

      args[3]="/test/testDistCh/sub3:sub1:sub2:447";
      newstatus[3] = new ChPermissionStatus(rootstatus, "sub1", "sub2", "447");
 
      args[4]="/test/testDistCh/sub4::sub5:437";
      newstatus[4] = new ChPermissionStatus(rootstatus, "", "sub5", "437");

      args[5]="/test/testDistCh/sub5:sub1:sub5:";
      newstatus[5] = new ChPermissionStatus(rootstatus, "sub1", "sub5", "");

      args[6]="/test/testDistCh/sub6:sub3::437";
      newstatus[6] = new ChPermissionStatus(rootstatus, "sub3", "", "437");
      
      System.out.println("args=" + Arrays.asList(args).toString().replace(",", ",\n  "));
      System.out.println("newstatus=" + Arrays.asList(newstatus).toString().replace(",", ",\n  "));

      //run DistCh
      new DistCh(MiniMRClientClusterFactory.create(this.getClass(), 2, conf).getConfig()).run(args);
      runLsr(shell, tree.root, 0);

      //check results
      for(int i = 0; i < NUN_SUBS; i++) {
        Path sub = new Path(tree.root + "/sub" + i);
        checkFileStatus(newstatus[i], fs.getFileStatus(sub));
        for(FileStatus status : fs.listStatus(sub)) {
          checkFileStatus(newstatus[i], status);
        }
      }
    } finally {
      cluster.shutdown();
    }
  }