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

        DistributedQueue<TestQueueItem> producerQueue = null;
        DistributedQueue<TestQueueItem> consumerQueue1 = null;
        DistributedQueue<TestQueueItem> consumerQueue2 = null;

        CuratorFramework producerClient =
                CuratorFrameworkFactory.newClient(server.getConnectString(), new RetryOneTime(1));
        CuratorFramework consumerClient1 =
                CuratorFrameworkFactory.newClient(server.getConnectString(), new RetryOneTime(1));
        CuratorFramework consumerClient2 =
                CuratorFrameworkFactory.newClient(server.getConnectString(), new RetryOneTime(1));
        try {
            producerClient.start();
            consumerClient1.start();
            consumerClient2.start();

            ExecutorService service = Executors.newCachedThreadPool();

            // make the producer queue
            {
                producerQueue = QueueBuilder.builder(producerClient, null, serializer, QUEUE_PATH)
                        .buildQueue();
                producerQueue.start();
                QueueTestProducer producer = new QueueTestProducer(producerQueue, itemQty, 0);
                service.submit(producer);
            }

            final Set<TestQueueItem> takenItems = Sets.newTreeSet();
            final Set<TestQueueItem> takenItemsForConsumer1 = Sets.newTreeSet();
            final Set<TestQueueItem> takenItemsForConsumer2 = Sets.newTreeSet();
            final AtomicReference<TestQueueItem> thrownItemFromConsumer1 = new AtomicReference<TestQueueItem>(null);

            // make the first consumer queue
            {
                final QueueConsumer<TestQueueItem> ourQueue = new QueueConsumer<TestQueueItem>() {
                    @Override
                    public void consumeMessage(TestQueueItem message) throws Exception {
                        synchronized (takenItems) {
                            if (takenItems.size() > 10) {
                                thrownItemFromConsumer1.set(message);
                                throw new Exception("dummy"); // simulate a crash
                            }
                        }

                        addToTakenItems(message, takenItems, itemQty);
                        synchronized (takenItemsForConsumer1) {
                            takenItemsForConsumer1.add(message);
                        }

                        Thread.sleep((long) (Math.random() * 5));
                    }

                    @Override
                    public void stateChanged(CuratorFramework client, ConnectionState newState) {}
                };
                consumerQueue1 = QueueBuilder.builder(consumerClient1, ourQueue, serializer, QUEUE_PATH)
                        .lockPath("/a/locks")
                        .buildQueue();
                consumerQueue1.start();
            }

            // make the second consumer queue
            {
                final QueueConsumer<TestQueueItem> ourQueue = new QueueConsumer<TestQueueItem>() {
                    @Override
                    public void consumeMessage(TestQueueItem message) throws Exception {
                        addToTakenItems(message, takenItems, itemQty);
                        synchronized (takenItemsForConsumer2) {
                            takenItemsForConsumer2.add(message);
                        }
                        Thread.sleep((long) (Math.random() * 5));
                    }

                    @Override
                    public void stateChanged(CuratorFramework client, ConnectionState newState) {}
                };
                consumerQueue2 = QueueBuilder.builder(consumerClient2, ourQueue, serializer, QUEUE_PATH)
                        .lockPath("/a/locks")
                        .buildQueue();
                consumerQueue2.start();
            }

            synchronized (takenItems) {
                while (takenItems.size() < itemQty) {
                    takenItems.wait(1000);
                }
            }

            int i = 0;
            for (TestQueueItem item : takenItems) {
                assertEquals(item.str, Integer.toString(i++));
            }

            assertNotNull(thrownItemFromConsumer1.get());
            assertTrue((takenItemsForConsumer2.contains(thrownItemFromConsumer1.get())));
            assertTrue(Sets.intersection(takenItemsForConsumer1, takenItemsForConsumer2)
                            .size()
                    == 0);
        } finally {
            CloseableUtils.closeQuietly(producerQueue);
            CloseableUtils.closeQuietly(consumerQueue1);
            CloseableUtils.closeQuietly(consumerQueue2);

            CloseableUtils.closeQuietly(producerClient);
            CloseableUtils.closeQuietly(consumerClient1);
            CloseableUtils.closeQuietly(consumerClient2);
        }
    }