@Test void skipsMalformedData() throws Exception {
    byte[] malformed1 = "[\"='".getBytes(UTF_8); // screwed up json
    byte[] malformed2 = "malformed".getBytes(UTF_8);
    pushMessage(collector.topic, THRIFT.encodeList(spans));
    pushMessage(collector.topic, new byte[0]);
    pushMessage(collector.topic, malformed1);
    pushMessage(collector.topic, malformed2);
    pushMessage(collector.topic, THRIFT.encodeList(spans));

    Thread.sleep(1000);

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