public <KIn, VIn, KOut, VOut> ProcessorSupplier<KIn, VIn, KOut, VOut> process(String spanName,
    ProcessorSupplier<KIn, VIn, KOut, VOut> processorSupplier) {
    return new TracingProcessorSupplier<>(this, spanName, processorSupplier);
  }
@Test void newProcessorSupplier_should_add_baggage_field() {
    ProcessorSupplier<String, String, String, String>
      processorSupplier = kafkaStreamsTracing.process("forward-1", () -> record ->
      assertThat(BAGGAGE_FIELD.getValue(currentTraceContext.get()))
        .isEqualTo("user1"));
    Headers headers = new RecordHeaders().add(BAGGAGE_FIELD_KEY, "user1".getBytes());
    Processor<String, String, String, String> processor = processorSupplier.get();
    processor.init(processorV2ContextSupplier.get());
    processor.process(new Record<>(TEST_KEY, TEST_VALUE, new Date().getTime(), headers));
  }