@Test
public void testNoOrphanedNodes() throws Exception
{

    final Timing timing = new Timing();
    final ExecutorService executor = Executors.newFixedThreadPool(1);
    CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), timing.session(), timing.connection(), new RetryOneTime(1));
    client.start();
    try
    {
        final InterProcessSemaphoreV2 semaphore = new InterProcessSemaphoreV2(client, "/test", 1);
        Lease lease = semaphore.acquire(timing.forWaiting().seconds(), TimeUnit.SECONDS);
        Assert.assertNotNull(lease);
        final List<String> childNodes = client.getChildren().forPath("/test/leases");
        Assert.assertEquals(childNodes.size(), 1);
        final CountDownLatch nodeCreatedLatch = new CountDownLatch(1);
        client.getChildren().usingWatcher(new CuratorWatcher()
        {
            @Override
            public void process(WatchedEvent event) throws Exception
            {
                if ( event.getType() == Watcher.Event.EventType.NodeCreated )
                {
                    nodeCreatedLatch.countDown();
                }
            }
        }).forPath("/test/leases");
        final Future<Lease> leaseFuture = executor.submit(new Callable<Lease>()
        {
            @Override
            public Lease call() throws Exception
            {
                return semaphore.acquire(timing.forWaiting().multiple(2).seconds(), TimeUnit.SECONDS);
            }
        });
        // wait for second lease to create its node
        timing.awaitLatch(nodeCreatedLatch);
        String newNode = null;
        for ( String c : client.getChildren().forPath("/test/leases") )
        {
            if ( !childNodes.contains(c) )
            {
                newNode = c;
            }
        }
        Assert.assertNotNull(newNode);
        // delete the ephemeral node to trigger a retry
        client.delete().forPath("/test/leases/" + newNode);
        // release first lease so second one can be acquired
        lease.close();
        lease = leaseFuture.get();
        Assert.assertNotNull(lease);
        lease.close();
        Assert.assertEquals(client.getChildren().forPath("/test/leases").size(), 0);
        // no more lease exist. must be possible to acquire a new one
        Assert.assertNotNull(semaphore.acquire(timing.forWaiting().seconds(), TimeUnit.SECONDS));
    }
    finally
    {
        executor.shutdownNow();
        TestCleanState.closeAndTestClean(client);
    }
}