AbstractFileOutputOperator extends BaseOperator implements Operator.CheckpointListener, Operator.CheckpointNotificationListener { public void setMaxOpenFiles(int maxOpenFiles) { this.maxOpenFiles = maxOpenFiles; } AbstractFileOutputOperator(); @Override void setup(Context.OperatorContext context); @Override void teardown(); @Override void beginWindow(long windowId); @Override void endWindow(); void setFilePath(@Nonnull String dir); String getFilePath(); void setMaxLength(long maxLength); long getMaxLength(); int getRotationWindows(); void setRotationWindows(int rotationWindows); void setMaxOpenFiles(int maxOpenFiles); int getMaxOpenFiles(); short getFilePermission(); void setFilePermission(short filePermission); FilterStreamProvider getFilterStreamProvider(); void setFilterStreamProvider(FilterStreamProvider filterStreamProvider); @Override void beforeCheckpoint(long l); @Override void checkpointed(long l); @Override void committed(long l); boolean isAlwaysWriteToTmp(); void setAlwaysWriteToTmp(boolean alwaysWriteToTmp); Long getExpireStreamAfterAccessMillis(); void setExpireStreamAfterAccessMillis(Long millis); static final int DEFAULT_MAX_OPEN_FILES; final transient DefaultInputPort<INPUT> input; }
@Test public void testMultiFileCompletedWriteCache1() { EvenOddHDFSExactlyOnceWriter writer = new EvenOddHDFSExactlyOnceWriter(); writer.setMaxOpenFiles(1); testMultiFileCompletedWriteHelper(writer); String evenFileName = testMeta.getDir() + File.separator + EVEN_FILE; String correctContents = "0\n" + "2\n" + "4\n" + "6\n"; checkOutput(-1, evenFileName, correctContents); String oddFileName = testMeta.getDir() + File.separator + ODD_FILE; correctContents = "1\n" + "3\n" + "5\n" + "7\n"; checkOutput(-1, oddFileName, correctContents); }
@Test public void testMultiFileCompletedWriteOverwriteCache1Initial() throws IOException { populateFile(EVEN_FILE, "0\n" + "2\n"); populateFile(ODD_FILE, "1\n" + "3\n"); EvenOddHDFSExactlyOnceWriter writer = new EvenOddHDFSExactlyOnceWriter(); writer.setMaxOpenFiles(1); testMultiFileCompletedWriteHelperCache1(writer); String evenFileName = testMeta.getDir() + File.separator + EVEN_FILE; String correctContents = "0\n" + "2\n" + "4\n" + "6\n"; checkOutput(-1, evenFileName, correctContents); String oddFileName = testMeta.getDir() + File.separator + ODD_FILE; correctContents = "1\n" + "3\n" + "5\n" + "7\n"; checkOutput(-1, oddFileName, correctContents); }
@Test public void testMultiFileFailedWriteCache1() { EvenOddHDFSExactlyOnceWriter writer = new EvenOddHDFSExactlyOnceWriter(); writer.setMaxOpenFiles(1); testMultiFileFailedWriteHelper(writer); String evenFileName = testMeta.getDir() + File.separator + EVEN_FILE; String correctContents = "0\n" + "2\n" + "6\n" + "8\n"; checkOutput(-1, evenFileName, correctContents); String oddFileName = testMeta.getDir() + File.separator + ODD_FILE; correctContents = "1\n" + "3\n" + "7\n" + "9\n"; checkOutput(-1, oddFileName, correctContents); }
@Test public void testMultiRollingFileCompletedWriteCache1() { EvenOddHDFSExactlyOnceWriter writer = new EvenOddHDFSExactlyOnceWriter(); writer.setMaxOpenFiles(1); testMultiRollingFileCompletedWriteHelper(writer); }
@Test public void testMultiRollingFileCompletedWriteOverwriteCache1() { EvenOddHDFSExactlyOnceWriter writer = new EvenOddHDFSExactlyOnceWriter(); writer.setMaxOpenFiles(1); testMultiRollingFileCompletedWriteHelperCache1(writer); }
@Test public void testMultiRollingFileFailedWriteCache1() { EvenOddHDFSExactlyOnceWriter writer = new EvenOddHDFSExactlyOnceWriter(); writer.setMaxOpenFiles(1); testMultiRollingFileFailedWriteHelperHelper(writer); }
@Test public void testMultiRollingFileFailedWriteOverwriteCache1() throws IOException { EvenOddHDFSExactlyOnceWriter writer = new EvenOddHDFSExactlyOnceWriter(); writer.setMaxOpenFiles(1); String evenFileName = testMeta.getDir() + File.separator + EVEN_FILE; String oddFileName = testMeta.getDir() + File.separator + ODD_FILE; populateFile(EVEN_FILE + ".0", "0\n" + "2\n" + "4\n"); populateFile(ODD_FILE + ".0", "1\n" + "3\n" + "5\n"); testMultiRollingFileFailedWriteOverwriteHelperCache1(writer); String correctContents = "0\n" + "4\n" + "6\n"; checkOutput(0, evenFileName, correctContents); correctContents = "8\n" + "6\n" + "10\n"; checkOutput(1, evenFileName, correctContents); correctContents = "1\n" + "5\n" + "7\n"; checkOutput(0, oddFileName, correctContents); correctContents = "9\n" + "7\n" + "11\n"; checkOutput(1, oddFileName, correctContents); }