@Test
public void testFastReplay() throws Exception {

  Map<String, String> overrides = Maps.newHashMap();
  overrides.put(FileChannelConfiguration.CAPACITY,
      String.valueOf(50));
  overrides.put(FileChannelConfiguration.TRANSACTION_CAPACITY,
      String.valueOf(50));
  channel = createFileChannel(overrides);
  channel.start();
  Assert.assertTrue(channel.isOpen());
  Set<String> in = fillChannel(channel, "checkpointBulder");
  channel.stop();
  File checkpointFile = new File(checkpointDir, "checkpoint");
  File metaDataFile = Serialization.getMetaDataFile(checkpointFile);
  File inflightTakesFile = new File(checkpointDir, "inflighttakes");
  File inflightPutsFile = new File(checkpointDir, "inflightputs");
  File queueSetDir = new File(checkpointDir, "queueset");
  Assert.assertTrue(checkpointFile.delete());
  Assert.assertTrue(metaDataFile.delete());
  Assert.assertTrue(inflightTakesFile.delete());
  Assert.assertTrue(inflightPutsFile.delete());
  EventQueueBackingStore backingStore =
      EventQueueBackingStoreFactory.get(checkpointFile, 50,
          "test", new FileChannelCounter("test"));
  FlumeEventQueue queue = new FlumeEventQueue(backingStore, inflightTakesFile,
        inflightPutsFile, queueSetDir);
  CheckpointRebuilder checkpointRebuilder =
      new CheckpointRebuilder(getAllLogs(dataDirs), queue, true);
  Assert.assertTrue(checkpointRebuilder.rebuild());
  channel = createFileChannel(overrides);
  channel.start();
  Assert.assertTrue(channel.isOpen());
  Set<String> out = consumeChannel(channel);
  compareInputAndOut(in, out);
}