@Test
public void testReferenceCounts() throws Exception {

  Map<String, String> overrides = Maps.newHashMap();
  overrides.put(FileChannelConfiguration.CHECKPOINT_INTERVAL, "10000");
  overrides.put(FileChannelConfiguration.MAX_FILE_SIZE, "150");
  final FileChannel channel = createFileChannel(overrides);
  channel.start();
  putEvents(channel, "testing-reference-counting", 1, 15);
  Transaction tx = channel.getTransaction();
  takeWithoutCommit(channel, tx, 10);
  forceCheckpoint(channel);
  tx.rollback();
  //Since we did not commit the original transaction. now we should get 15
  //events back.
  final Set<String> takenEvents = Sets.newHashSet();
  Executors.newSingleThreadExecutor().submit(new Runnable() {
    @Override
    public void run() {
      try {
        takenEvents.addAll(takeEvents(channel, 15));
      } catch (Exception ex) {
        Throwables.propagate(ex);
      }
    }
  }).get();
  Assert.assertEquals(15, takenEvents.size());
}