public Span nextSpan(ConsumerRecord<?, ?> record) {
    // Even though the type is ConsumerRecord, this is not a (remote) consumer span. Only "poll"
    // events create consumer spans. Since this is a processor span, we use the normal sampler.
    TraceContextOrSamplingFlags extracted =
      extractAndClearTraceIdHeaders(processorExtractor, record.headers(), record.headers());
    Span result = tracer.nextSpan(extracted);
    if (extracted.context() == null && !result.isNoop()) {
      addTags(record, result);
    }
    return result;
  }
@Test void nextSpan_shouldnt_tag_null_key() {
    consumerRecord = new ConsumerRecord<>(TEST_TOPIC, 0, 1, null, TEST_VALUE);

    kafkaTracing.nextSpan(consumerRecord).start().finish();

    assertThat(spans.get(0).tags())
      .containsOnly(entry("kafka.topic", TEST_TOPIC));
  }