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

    AtomicInteger counter = new AtomicInteger();
    consumer = (input) -> new Call.Base<>() {
      @Override protected Void doExecute() {
        throw new AssertionError();
      }

      @Override protected void doEnqueue(Callback<Void> callback) {
        if (counter.getAndIncrement() == 1) {
          callback.onError(new RuntimeException("storage fell over"));
        } else {
          receivedSpans.add(spans);
          callback.onSuccess(null);
        }
      }

      @Override public Call<Void> clone() {
        throw new AssertionError();
      }
    };

    collector = builder().storage(buildStorage(consumer)).build().start();

    pushMessage(collector.topic, PROTO3.encodeList(spans));
    pushMessage(collector.topic, PROTO3.encodeList(spans)); // tossed on error
    pushMessage(collector.topic, PROTO3.encodeList(spans));

    assertThat(receivedSpans.take()).containsExactlyElementsOf(spans);
    // the only way we could read this, is if the malformed span was skipped.
    assertThat(receivedSpans.take()).containsExactlyElementsOf(spans);

    assertThat(pulsarMetrics.messages()).isEqualTo(3);
    assertThat(pulsarMetrics.messagesDropped()).isZero(); // storage failure not message failure
    assertThat(pulsarMetrics.bytes()).isEqualTo(PROTO3.encodeList(spans).length * 3);
    assertThat(pulsarMetrics.spans()).isEqualTo(spans.size() * 3);
    assertThat(pulsarMetrics.spansDropped()).isEqualTo(spans.size()); // only one dropped
  }
Guards against errors that leak from storage, such as InvalidQueryException
zipkin-collector/pulsar/src/test/java/zipkin2/collector/pulsar/ITPulsarCollector.java