gistfile1.java

    @Test
    public void testRetriesExpireAndReconnect() throws Exception
    {
        Timing timing = new Timing();
        PersistentEphemeralNode node = null;
        CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), timing.session(), timing.connection(), new RetryOneTime(1));
        try
        {
            client.start();
            node = new PersistentEphemeralNode(client, PersistentEphemeralNode.Mode.EPHEMERAL, "/abc/node", "hello".getBytes());
            node.start();
            Assert.assertTrue(node.waitForInitialCreate(timing.forWaiting().seconds(), TimeUnit.SECONDS));

            final CountDownLatch lostLatch = new CountDownLatch(1);
            ConnectionStateListener listener = new ConnectionStateListener()
            {
                @Override
                public void stateChanged(CuratorFramework client, ConnectionState newState)
                {
                    if ( newState == ConnectionState.LOST )
                    {
                        lostLatch.countDown();
                    }
                }
            };
            client.getConnectionStateListenable().addListener(listener);

            server.stop();

            Assert.assertTrue(timing.awaitLatch(lostLatch));
            CountDownLatch recreateLatch = new CountDownLatch(1);
            node.initialCreateLatch.set(recreateLatch);

            TimeUnit.MILLISECONDS.sleep(2 * timing.session());  // make sure session expires

            server = new TestingServer(server.getPort(), server.getTempDirectory());
            timing.sleepABit();
            KillSession.kill(client.getZookeeperClient().getZooKeeper(), server.getConnectString());
            timing.sleepABit();
            Assert.assertTrue(timing.awaitLatch(recreateLatch));
            Assert.assertNotNull(client.checkExists().forPath("/abc/node"));
        }
        finally
        {
            CloseableUtils.closeQuietly(node);
            CloseableUtils.closeQuietly(client);
        }
    }