FSRecordCompactionOperator extends GenericFileOutputOperator<INPUT> { @Override public void beginWindow(long windowId) { super.beginWindow(windowId); while (!emitQueue.isEmpty()) { output.emit(emitQueue.poll()); } } FSRecordCompactionOperator(); @Override void setup(Context.OperatorContext context); @Override void beginWindow(long windowId); String getOutputDirectoryName(); void setOutputDirectoryName(@NotNull String outputDirectoryName); String getOutputFileNamePrefix(); void setOutputFileNamePrefix(@NotNull String outputFileNamePrefix); transient DefaultOutputPort<OutputMetaData> output; }
@Test public void testRotate() throws Exception { CollectorTestSink<FSRecordCompactionOperator.OutputMetaData> sink = new CollectorTestSink<FSRecordCompactionOperator.OutputMetaData>(); testMeta.underTest.output.setSink((CollectorTestSink)sink); for (int i = 0; i < 60; i++) { testMeta.underTest.beginWindow(i); if (i < 10) { testMeta.underTest.input.process(("Record" + Integer.toString(i)).getBytes()); } testMeta.underTest.endWindow(); } testMeta.underTest.committed(59); for (int i = 60; i < 70; i++) { testMeta.underTest.beginWindow(i); testMeta.underTest.endWindow(); } Assert.assertEquals("tuples-" + testMeta.context.getAttributes().get(DAG.DAGContext.APPLICATION_ID) + "_1.0", sink.collectedTuples.get(0).getFileName()); }