@Test void skipsMalformedData() throws Exception {
    KafkaCollector.Builder builder = builder("decoder_exception");

    byte[] malformed1 = "[\"='".getBytes(UTF_8); // screwed up json
    byte[] malformed2 = "malformed".getBytes(UTF_8);
    produceSpans(THRIFT.encodeList(spans), builder.topic);
    produceSpans(new byte[0], builder.topic);
    produceSpans(malformed1, builder.topic);
    produceSpans(malformed2, builder.topic);
    produceSpans(THRIFT.encodeList(spans), builder.topic);

    try (KafkaCollector collector = builder.build()) {
      collector.start();
      assertThat(receivedSpans.take()).containsExactlyElementsOf(spans);
      // the only way we could read this, is if the malformed spans were skipped.
      assertThat(receivedSpans.take()).containsExactlyElementsOf(spans);
    }

    assertThat(kafkaMetrics.messages()).isEqualTo(5);
    assertThat(kafkaMetrics.messagesDropped()).isEqualTo(2); // only malformed, not empty
    assertThat(kafkaMetrics.bytes())
      .isEqualTo(THRIFT.encodeList(spans).length * 2 + malformed1.length + malformed2.length);
    assertThat(kafkaMetrics.spans()).isEqualTo(spans.size() * 2);
    assertThat(kafkaMetrics.spansDropped()).isZero();
  }
Ensures malformed spans don't hang the collector
zipkin-collector/kafka/src/test/java/zipkin2/collector/kafka/ITKafkaCollector.java