FileChannel extends BasicChannelSemantics implements TransactionCapacitySupported { @Override public synchronized void start() { LOG.info("Starting {}...", this); channelCounter.start(); try { Builder builder = createLogBuilder(); log = builder.build(); log.replay(); setOpen(true); int depth = getDepth(); Preconditions.checkState(queueRemaining.tryAcquire(depth), "Unable to acquire " + depth + " permits " + channelNameDescriptor); LOG.info("Queue Size after replay: " + depth + " " + channelNameDescriptor); } catch (Throwable t) { setOpen(false); channelCounter.setUnhealthy(1); startupError = t; LOG.error("Failed to start the file channel " + channelNameDescriptor, t); if (t instanceof Error) { throw (Error) t; } } if (open) { channelCounter.setChannelSize(getDepth()); channelCounter.setChannelCapacity(capacity); } super.start(); }  @Override synchronized void setName(String name); @Override void configure(Context context); @Override synchronized void start(); @Override synchronized void stop(); @Override String toString(); boolean isOpen(); @Override long getTransactionCapacity();  }
@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(); 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()); }