AvroFileInputOperator extends AbstractFileInputOperator<GenericRecord> { @Override public void beginWindow(long windowId) { super.beginWindow(windowId); errorCount = 0; recordCount = 0; }  @Override void beginWindow(long windowId); final transient DefaultOutputPort<GenericRecord> output; final transient DefaultOutputPort<String> completedFilesPort; final transient DefaultOutputPort<String> errorRecordsPort; }
@Test public void testSingleFileAvroReads() throws Exception { FileContext.getLocalFSFileContext().delete(new Path(new File(testMeta.dir).getAbsolutePath()), true); int cnt = 7; createAvroInput(cnt); writeAvroFile(new File(FILENAME)); 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.emitTuples(); Assert.assertEquals("Record count", cnt, avroFileInput.recordCount); avroFileInput.endWindow(); Assert.assertEquals("number tuples", cnt, output.collectedTuples.size()); Assert.assertEquals("Error tuples", 0, errorRecordsPort.collectedTuples.size()); Assert.assertEquals("Completed File", 1, completedFilesPort.collectedTuples.size()); avroFileInput.teardown(); }
@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(); }
@Test public void testInvalidFormatFailure() throws Exception { FileContext.getLocalFSFileContext().delete(new Path(new File(testMeta.dir).getAbsolutePath()), true); int cnt = 7; writeErrorFile(cnt, new File(ERROR_FILE)); avroFileInput.output.setSink(output); avroFileInput.setDirectory(testMeta.dir); avroFileInput.setup(testMeta.context); avroFileInput.beginWindow(0); avroFileInput.emitTuples(); avroFileInput.emitTuples(); avroFileInput.endWindow(); Assert.assertEquals("number tuples after window 1", 0, output.collectedTuples.size()); avroFileInput.teardown(); }