NioForceSynchronizer { public boolean waitOnEnlisted() { long enlistedElementNumber = pendingRecordsQueue.getMaxElementSequenceNumberForCurrentThread(true); try { forceLock.lockInterruptibly(); try { while (enlistedElementNumber > latestForcedElement.get()) { if (verifyIsInFailedRange(enlistedElementNumber)) return false; if (log.isDebugEnabled()) { log.debug("Waiting until entry with sequence " + enlistedElementNumber + " was forced."); } performedForce.await(); } } finally { forceLock.unlock(); } } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new RuntimeException(e); } if (verifyIsInFailedRange(enlistedElementNumber)) { return false; } else { if (log.isDebugEnabled()) { log.debug("Entry with sequence " + enlistedElementNumber + " was successfully forced (force ranges up to " + latestForcedElement.get() + ")."); } return true; } } NioForceSynchronizer(SequencedBlockingQueue pendingRecordsQueue); boolean waitOnEnlisted(); int getNumberOfWaitingThreads(); boolean processEnlistedIfRequired(Callable forceCommand,
                                             Collection<? extends SequencedQueueEntry> elements); void processEnlisted(Callable forceCommand,
                                Collection<? extends SequencedQueueEntry> elements);  }
@Test public void testWaitOnEnlisted() throws Exception { List<Future<Boolean>> futures = doTestWaitOnEnlistedWithSuccess(); for (Future<Boolean> future : futures) assertTrue(future.get()); }