@Test
public void testSequenceFileFromDirectoryBasic() throws Exception {

  // parameters
  Configuration configuration = getConfiguration();
  FileSystem fs = FileSystem.get(configuration);
  // create
  Path tmpDir = this.getTestTempDirPath();
  Path inputDir = new Path(tmpDir, "inputDir");
  fs.mkdirs(inputDir);
  Path outputDir = new Path(tmpDir, "outputDir");
  Path outputDirRecursive = new Path(tmpDir, "outputDirRecursive");
  Path inputDirRecursive = new Path(tmpDir, "inputDirRecur");
  fs.mkdirs(inputDirRecursive);
  // prepare input files
  createFilesFromArrays(configuration, inputDir, DATA1);
  SequenceFilesFromDirectory.main(new String[]{
    "--input", inputDir.toString(),
    "--output", outputDir.toString(),
    "--chunkSize", "64",
    "--charset", Charsets.UTF_8.name(),
    "--keyPrefix", "UID",
    "--method", "sequential"});
  // check output chunk files
  checkChunkFiles(configuration, outputDir, DATA1, "UID");
  createRecursiveDirFilesFromArrays(configuration, inputDirRecursive, DATA2);
  FileStatus fstInputPath = fs.getFileStatus(inputDirRecursive);
  String dirs = HadoopUtil.buildDirList(fs, fstInputPath);
  System.out.println("\n\n ----- recursive dirs: " + dirs);
  SequenceFilesFromDirectory.main(new String[]{
    "--input", inputDirRecursive.toString(),
    "--output", outputDirRecursive.toString(),
    "--chunkSize", "64",
    "--charset", Charsets.UTF_8.name(),
    "--keyPrefix", "UID",
    "--method", "sequential"});
  checkRecursiveChunkFiles(configuration, outputDirRecursive, DATA2, "UID");
}