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_should_create_span_with_baggage() {
    addB3MultiHeaders(parent, consumerRecord);
    consumerRecord.headers().add(BAGGAGE_FIELD_KEY, "user1".getBytes());

    Span span = kafkaTracing.nextSpan(consumerRecord);
    assertThat(BAGGAGE_FIELD.getValue(span.context())).contains("user1");
  }