@Test
public void testMultipleFileAvroReads() throws Exception
{

  FileContext.getLocalFSFileContext().delete(new Path(new File(testMeta.dir).getAbsolutePath()), true);
  int cnt = 7;
  createAvroInput(cnt);
  writeAvroFile(new File(FILENAME));
  writeAvroFile(new File(OTHER_FILE));
  avroFileInput.output.setSink(output);
  avroFileInput.completedFilesPort.setSink(completedFilesPort);
  avroFileInput.errorRecordsPort.setSink(errorRecordsPort);
  avroFileInput.setDirectory(testMeta.dir);
  avroFileInput.setup(testMeta.context);
  avroFileInput.beginWindow(0);
  avroFileInput.emitTuples();
  avroFileInput.beginWindow(1);
  avroFileInput.emitTuples();
  Assert.assertEquals("number tuples after window 0", cnt, output.collectedTuples.size());
  avroFileInput.emitTuples();
  avroFileInput.endWindow();
  Assert.assertEquals("Error tuples", 0, errorRecordsPort.collectedTuples.size());
  Assert.assertEquals("number tuples after window 1", 2 * cnt, output.collectedTuples.size());
  Assert.assertEquals("Completed File", 2, completedFilesPort.collectedTuples.size());
  avroFileInput.teardown();
}