@Test
public void testTokenizeDocuments() throws Exception {

  Configuration configuration = new Configuration();
  Path input = new Path(getTestTempDirPath(), "inputDir");
  Path output = new Path(getTestTempDirPath(), "outputDir");
  FileSystem fs = FileSystem.get(input.toUri(), configuration);
  String documentId1 = "123";
  String documentId2 = "456";
  SequenceFile.Writer writer = new SequenceFile.Writer(fs, configuration, input, Text.class, Text.class);
  try {
    String text1 = "A test for the document processor";
    writer.append(new Text(documentId1), new Text(text1));
    String text2 = "and another one";
    writer.append(new Text(documentId2), new Text(text2));
  } finally {
    Closeables.closeQuietly(writer);
  }
  DocumentProcessor.tokenizeDocuments(input, StandardAnalyzer.class, output, configuration);
  FileStatus[] statuses = fs.listStatus(output, PathFilters.logsCRCFilter());
  assertEquals(1, statuses.length);
  Path filePath = statuses[0].getPath();
  SequenceFile.Reader reader = new SequenceFile.Reader(fs, filePath, configuration);
  Text key = ClassUtils.instantiateAs((Class<? extends Text>) reader.getKeyClass(), Text.class);
  StringTuple value =
      ClassUtils.instantiateAs((Class<? extends StringTuple>) reader.getValueClass(), StringTuple.class);
  reader.next(key, value);
  assertEquals(documentId1, key.toString());
  assertEquals(Arrays.asList("test", "document", "processor"), value.getEntries());
  reader.next(key, value);
  assertEquals(documentId2, key.toString());
  assertEquals(Arrays.asList("another", "one"), value.getEntries());
}