@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());
}