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_prefers_b3_header() {
    consumerRecord.headers().add("b3", B3SingleFormat.writeB3SingleFormatAsBytes(incoming));

    Span child;
    try (Scope scope = tracing.currentTraceContext().newScope(parent)) {
      child = kafkaTracing.nextSpan(consumerRecord);
    }
    child.finish();

    assertThat(spans.get(0).id()).isEqualTo(child.context().spanIdString());
    assertChildOf(spans.get(0), incoming);
  }