  @Test
  public void skipsOnConsumerException() throws Exception {

    KafkaCollector.Builder builder = builder("consumer_exception");

    AtomicInteger counter = new AtomicInteger();

    consumer = (spans, callback) -> {
      if (counter.getAndIncrement() == 1) {
        callback.onError(new RuntimeException("storage fell over"));
      } else {
        recvdSpans.add(spans);
        callback.onSuccess(null);
      }
    };

    producer.send(new KeyedMessage<>(builder.topic, Codec.JSON.writeSpans(TRACE)));
    producer.send(new KeyedMessage<>(builder.topic, Codec.JSON.writeSpans(TRACE))); // tossed on error
    producer.send(new KeyedMessage<>(builder.topic, Codec.JSON.writeSpans(TRACE)));

    try (KafkaCollector collector = newKafkaTransport(builder, consumer)) {
      assertThat(recvdSpans.take()).containsExactlyElementsOf(TRACE);
      // the only way we could read this, is if the malformed span was skipped.
      assertThat(recvdSpans.take()).containsExactlyElementsOf(TRACE);
    }

    assertThat(kafkaMetrics.spansDropped()).isEqualTo(TestObjects.TRACE.size());
  }