@Override
    public void start() throws Exception {
        if (!state.compareAndSet(State.LATENT, State.STARTED)) {
            throw new IllegalStateException();
        }

        try {
            client.create().creatingParentContainersIfNeeded().forPath(queuePath);
        } catch (KeeperException.NodeExistsException ignore) {
            // this is OK
        }
        if (lockPath != null) {
            try {
                client.create().creatingParentContainersIfNeeded().forPath(lockPath);
            } catch (KeeperException.NodeExistsException ignore) {
                // this is OK
            }
        }

        if (!isProducerOnly || (maxItems != QueueBuilder.NOT_SET)) {
            childrenCache.start();
        }

        if (!isProducerOnly) {
            service.submit(new Callable<Object>() {
                @Override
                public Object call() {
                    runLoop();
                    return null;
                }
            });
        }
    }
@Test
    public void testSafetyBasic() throws Exception {
        final int itemQty = 10;

        DistributedQueue<TestQueueItem> queue = null;
        CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), new RetryOneTime(1));
        client.start();
        try {
            final BlockingQueueConsumer<TestQueueItem> consumer =
                    new BlockingQueueConsumer<>(new DummyConnectionStateListener());
            queue = QueueBuilder.builder(client, consumer, serializer, QUEUE_PATH)
                    .lockPath("/a/locks")
                    .buildQueue();
            queue.start();

            QueueTestProducer producer = new QueueTestProducer(queue, itemQty, 0);

            ExecutorService service = Executors.newCachedThreadPool();
            service.submit(producer);

            final CountDownLatch latch = new CountDownLatch(1);
            service.submit(new Callable<Object>() {
                @Override
                public Object call() throws Exception {
                    for (int i = 0; i < itemQty; ++i) {
                        TestQueueItem item = consumer.take();
                        assertEquals(item.str, Integer.toString(i));
                    }
                    latch.countDown();
                    return null;
                }
            });
            assertTrue(latch.await(10, TimeUnit.SECONDS));
        } finally {
            CloseableUtils.closeQuietly(queue);
            CloseableUtils.closeQuietly(client);
        }
    }