@Test void messagesDistributedAcrossMultipleThreadsSuccessfully() throws Exception {
    collector.close();

    CountDownLatch latch = new CountDownLatch(2);
    collector = builder().concurrency(2).storage(buildStorage((spans) -> {
      latch.countDown();
      try {
        latch.await(); // await the other one as this proves 2 threads are in use
      } catch (InterruptedException e) {
        throw new AssertionError(e);
      }
      return consumer.accept(spans);
    })).build().start();

    pushMessage(collector.topic, new byte[]{}); // empty bodies don't go to storage
    pushMessage(collector.topic, PROTO3.encodeList(spans));
    pushMessage(collector.topic, PROTO3.encodeList(spans));

    assertThat(receivedSpans.take()).containsExactlyElementsOf(spans);
    latch.countDown();
    assertThat(receivedSpans.take()).containsExactlyElementsOf(spans);

    assertThat(threadsProvidingSpans).hasSize(2);

    assertThat(pulsarMetrics.messages()).isEqualTo(3); // 2 + empty body for warmup
    assertThat(pulsarMetrics.messagesDropped()).isZero();
    assertThat(pulsarMetrics.bytes()).isEqualTo(PROTO3.encodeList(spans).length * 2);
    assertThat(pulsarMetrics.spans()).isEqualTo(spans.size() * 2);
    assertThat(pulsarMetrics.spansDropped()).isZero();
  }
Guards against errors that leak from storage, such as InvalidQueryException
zipkin-collector/pulsar/src/test/java/zipkin2/collector/pulsar/ITPulsarCollector.java