@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()); }
@Override public synchronized void start() { LOG.info("Starting {}...", this); try { Builder builder = new Log.Builder(); builder.setCheckpointInterval(checkpointInterval); builder.setMaxFileSize(maxFileSize); builder.setMinimumRequiredSpace(minimumRequiredSpace); builder.setQueueSize(capacity); builder.setLogWriteTimeout(logWriteTimeout); builder.setCheckpointDir(checkpointDir); builder.setLogDirs(dataDirs); builder.setChannelName(getName()); builder.setCheckpointWriteTimeout(checkpointWriteTimeout); builder.setUseLogReplayV1(useLogReplayV1); builder.setUseFastReplay(useFastReplay); builder.setEncryptionKeyProvider(encryptionKeyProvider); builder.setEncryptionKeyAlias(encryptionActiveKey); builder.setEncryptionCipherProvider(encryptionCipherProvider); builder.setUseDualCheckpoints(useDualCheckpoints); builder.setBackupCheckpointDir(backupCheckpointDir); log = builder.build(); log.replay(); open = 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) { open = false; startupError = t; LOG.error("Failed to start the file channel " + channelNameDescriptor, t); if (t instanceof Error) { throw (Error) t; } } if (open) { channelCounter.start(); channelCounter.setChannelSize(getDepth()); channelCounter.setChannelCapacity(capacity); } super.start(); }
FileChannel extends BasicChannelSemantics { @Override public synchronized void start() { LOG.info("Starting {}...", this); try { Builder builder = new Log.Builder(); builder.setCheckpointInterval(checkpointInterval); builder.setMaxFileSize(maxFileSize); builder.setMinimumRequiredSpace(minimumRequiredSpace); builder.setQueueSize(capacity); builder.setLogWriteTimeout(logWriteTimeout); builder.setCheckpointDir(checkpointDir); builder.setLogDirs(dataDirs); builder.setChannelName(getName()); builder.setCheckpointWriteTimeout(checkpointWriteTimeout); builder.setUseLogReplayV1(useLogReplayV1); builder.setUseFastReplay(useFastReplay); builder.setEncryptionKeyProvider(encryptionKeyProvider); builder.setEncryptionKeyAlias(encryptionActiveKey); builder.setEncryptionCipherProvider(encryptionCipherProvider); builder.setUseDualCheckpoints(useDualCheckpoints); builder.setBackupCheckpointDir(backupCheckpointDir); log = builder.build(); log.replay(); open = 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) { open = false; startupError = t; LOG.error("Failed to start the file channel " + channelNameDescriptor, t); if (t instanceof Error) { throw (Error) t; } } if (open) { channelCounter.start(); channelCounter.setChannelSize(getDepth()); channelCounter.setChannelCapacity(capacity); } super.start(); } }
FileChannel extends BasicChannelSemantics { @Override public synchronized void start() { LOG.info("Starting {}...", this); try { Builder builder = new Log.Builder(); builder.setCheckpointInterval(checkpointInterval); builder.setMaxFileSize(maxFileSize); builder.setMinimumRequiredSpace(minimumRequiredSpace); builder.setQueueSize(capacity); builder.setLogWriteTimeout(logWriteTimeout); builder.setCheckpointDir(checkpointDir); builder.setLogDirs(dataDirs); builder.setChannelName(getName()); builder.setCheckpointWriteTimeout(checkpointWriteTimeout); builder.setUseLogReplayV1(useLogReplayV1); builder.setUseFastReplay(useFastReplay); builder.setEncryptionKeyProvider(encryptionKeyProvider); builder.setEncryptionKeyAlias(encryptionActiveKey); builder.setEncryptionCipherProvider(encryptionCipherProvider); builder.setUseDualCheckpoints(useDualCheckpoints); builder.setBackupCheckpointDir(backupCheckpointDir); log = builder.build(); log.replay(); open = 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) { open = false; startupError = t; LOG.error("Failed to start the file channel " + channelNameDescriptor, t); if (t instanceof Error) { throw (Error) t; } } if (open) { channelCounter.start(); channelCounter.setChannelSize(getDepth()); channelCounter.setChannelCapacity(capacity); } super.start(); }  }
FileChannel extends BasicChannelSemantics { @Override public synchronized void start() { LOG.info("Starting {}...", this); try { Builder builder = new Log.Builder(); builder.setCheckpointInterval(checkpointInterval); builder.setMaxFileSize(maxFileSize); builder.setMinimumRequiredSpace(minimumRequiredSpace); builder.setQueueSize(capacity); builder.setLogWriteTimeout(logWriteTimeout); builder.setCheckpointDir(checkpointDir); builder.setLogDirs(dataDirs); builder.setChannelName(getName()); builder.setCheckpointWriteTimeout(checkpointWriteTimeout); builder.setUseLogReplayV1(useLogReplayV1); builder.setUseFastReplay(useFastReplay); builder.setEncryptionKeyProvider(encryptionKeyProvider); builder.setEncryptionKeyAlias(encryptionActiveKey); builder.setEncryptionCipherProvider(encryptionCipherProvider); builder.setUseDualCheckpoints(useDualCheckpoints); builder.setBackupCheckpointDir(backupCheckpointDir); log = builder.build(); log.replay(); open = 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) { open = false; startupError = t; LOG.error("Failed to start the file channel " + channelNameDescriptor, t); if (t instanceof Error) { throw (Error) t; } } if (open) { channelCounter.start(); 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(); }
FileChannel extends BasicChannelSemantics { @Override public synchronized void start() { LOG.info("Starting {}...", this); try { Builder builder = new Log.Builder(); builder.setCheckpointInterval(checkpointInterval); builder.setMaxFileSize(maxFileSize); builder.setMinimumRequiredSpace(minimumRequiredSpace); builder.setQueueSize(capacity); builder.setLogWriteTimeout(logWriteTimeout); builder.setCheckpointDir(checkpointDir); builder.setLogDirs(dataDirs); builder.setChannelName(getName()); builder.setCheckpointWriteTimeout(checkpointWriteTimeout); builder.setUseLogReplayV1(useLogReplayV1); builder.setUseFastReplay(useFastReplay); builder.setEncryptionKeyProvider(encryptionKeyProvider); builder.setEncryptionKeyAlias(encryptionActiveKey); builder.setEncryptionCipherProvider(encryptionCipherProvider); builder.setUseDualCheckpoints(useDualCheckpoints); builder.setBackupCheckpointDir(backupCheckpointDir); log = builder.build(); log.replay(); open = 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) { open = false; startupError = t; LOG.error("Failed to start the file channel " + channelNameDescriptor, t); if (t instanceof Error) { throw (Error) t; } } if (open) { channelCounter.start(); 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();  }