@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 testMultiPutterSingleGetter() throws Exception {
        final int itemQty = 100;

        DistributedQueue<TestQueueItem> queue = null;
        CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), new RetryOneTime(1));
        client.start();
        try {
            BlockingQueueConsumer<TestQueueItem> consumer =
                    new BlockingQueueConsumer<>(new DummyConnectionStateListener());

            queue = QueueBuilder.builder(client, consumer, serializer, QUEUE_PATH)
                    .buildQueue();
            queue.start();

            QueueTestProducer producer1 = new QueueTestProducer(queue, itemQty / 2, 0);
            QueueTestProducer producer2 = new QueueTestProducer(queue, ((itemQty + 1) / 2), itemQty / 2);

            ExecutorService service = Executors.newCachedThreadPool();
            service.submit(producer1);
            service.submit(producer2);

            int iteration = 0;
            while (consumer.size() < itemQty) {
                assertTrue(++iteration < 10);
                Thread.sleep(1000);
            }

            List<TestQueueItem> items = consumer.getItems();

            assertEquals(
                    com.google.common.collect.Sets.<TestQueueItem>newHashSet(items)
                            .size(),
                    items.size()); // check no duplicates
        } finally {
            CloseableUtils.closeQuietly(queue);
            CloseableUtils.closeQuietly(client);
        }
    }