@Test void messagesDistributedAcrossMultipleThreadsSuccessfully() throws Exception {
    KafkaCollector.Builder builder = builder("multi_thread", 2);

    kafka.prepareTopics(builder.topic, 2);
    warmUpTopic(builder.topic);

    final byte[] traceBytes = JSON_V2.encodeList(spans);
    try (KafkaCollector collector = builder.build()) {
      collector.start();
      waitForPartitionAssignments(collector);
      produceSpans(traceBytes, builder.topic, 0);
      assertThat(receivedSpans.take()).containsExactlyElementsOf(spans);
      produceSpans(traceBytes, builder.topic, 1);
      assertThat(receivedSpans.take()).containsExactlyElementsOf(spans);
    }

    assertThat(threadsProvidingSpans).hasSize(2);

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