public void start() {
        Preconditions.checkState(state.compareAndSet(State.LATENT, State.STARTED), "Cannot be started more than once");

        service.submit(new Callable<Object>() {
            @Override
            public Object call() throws Exception {
                processEvents();
                return null;
            }
        });
    }
@Test
    @Tag(CuratorTestBase.zk36Group)
    public void testConnectionStateRecoversFromUnexpectedExpiredConnection() throws Exception {
        Timing2 timing = new Timing2();
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString(server.getConnectString())
                .connectionTimeoutMs(1_000)
                .sessionTimeoutMs(250) // try to aggressively expire the connection
                .retryPolicy(new RetryOneTime(1))
                .connectionStateErrorPolicy(new SessionConnectionStateErrorPolicy())
                .build();
        final BlockingQueue<ConnectionState> queue = Queues.newLinkedBlockingQueue();
        ConnectionStateListener listener = (client1, state) -> queue.add(state);
        client.getConnectionStateListenable().addListener(listener);
        client.start();
        try {
            ConnectionState polled = queue.poll(timing.forWaiting().seconds(), TimeUnit.SECONDS);
            assertEquals(polled, ConnectionState.CONNECTED);
            client.getZookeeperClient()
                    .getZooKeeper()
                    .getTestable()
                    .queueEvent(new WatchedEvent(
                            Watcher.Event.EventType.None, Watcher.Event.KeeperState.Disconnected, null));
            polled = queue.poll(timing.forWaiting().seconds(), TimeUnit.SECONDS);
            assertEquals(polled, ConnectionState.SUSPENDED);
            assertThrows(RuntimeException.class, () -> client.getZookeeperClient()
                    .getZooKeeper()
                    .getTestable()
                    .queueEvent(
                            new WatchedEvent(Watcher.Event.EventType.None, Watcher.Event.KeeperState.Expired, null) {
                                @Override
                                public String getPath() {
                                    // exception will cause ZooKeeper to update current state but fail to notify
                                    // watchers
                                    throw new RuntimeException("Path doesn't exist!");
                                }
                            }));
            polled = queue.poll(timing.forWaiting().seconds(), TimeUnit.SECONDS);
            assertEquals(polled, ConnectionState.LOST);
            polled = queue.poll(timing.forWaiting().seconds(), TimeUnit.SECONDS);
            assertEquals(polled, ConnectionState.RECONNECTED);
        } finally {
            CloseableUtils.closeQuietly(client);
        }
    }