KafkaStreamsTracing { public <K, V> ProcessorSupplier<K, V> processor(String spanName, ProcessorSupplier<K, V> processorSupplier) { return new TracingProcessorSupplier<>(this, spanName, processorSupplier); } KafkaStreamsTracing(Builder builder); static KafkaStreamsTracing create(Tracing tracing); static KafkaStreamsTracing create(MessagingTracing messagingTracing); static KafkaStreamsTracing create(KafkaTracing kafkaTracing); static Builder newBuilder(Tracing tracing); static Builder newBuilder(MessagingTracing messagingTracing); KafkaClientSupplier kafkaClientSupplier(); KafkaStreams kafkaStreams(Topology topology, Properties streamsConfig); ProcessorSupplier<K, V> processor(String spanName,
    ProcessorSupplier<K, V> processorSupplier); TransformerSupplier<K, V, R> transformer(String spanName,
    TransformerSupplier<K, V, R> transformerSupplier); ValueTransformerSupplier<V, VR> valueTransformer(String spanName,
    ValueTransformerSupplier<V, VR> valueTransformerSupplier); ValueTransformerWithKeySupplier<K, V, VR> valueTransformerWithKey(
    String spanName,
    ValueTransformerWithKeySupplier<K, V, VR> valueTransformerWithKeySupplier); ProcessorSupplier<K, V> foreach(String spanName, ForeachAction<K, V> action); ValueTransformerWithKeySupplier<K, V, V> peek(String spanName,
    ForeachAction<K, V> action); ValueTransformerWithKeySupplier<K, V, V> mark(String spanName); TransformerSupplier<K, V, KeyValue<KR, VR>> map(String spanName,
    KeyValueMapper<K, V, KeyValue<KR, VR>> mapper); TransformerSupplier<K, V, Iterable<KeyValue<KR, VR>>> flatMap(
    String spanName,
    KeyValueMapper<K, V, Iterable<KeyValue<KR, VR>>> mapper); TransformerSupplier<K, V, KeyValue<K, V>> filter(String spanName,
    Predicate<K, V> predicate); TransformerSupplier<K, V, KeyValue<K, V>> filterNot(String spanName,
    Predicate<K, V> predicate); ValueTransformerWithKeySupplier<K, V, V> markAsFiltered(String spanName,
    Predicate<K, V> predicate); ValueTransformerWithKeySupplier<K, V, V> markAsNotFiltered(String spanName,
    Predicate<K, V> predicate); ValueTransformerWithKeySupplier<K, V, VR> mapValues(String spanName,
    ValueMapperWithKey<K, V, VR> mapper); ValueTransformerSupplier<V, VR> mapValues(String spanName,
    ValueMapper<V, VR> mapper);  }
@Test public void processorSupplier_should_add_baggage_field() { ProcessorSupplier<String, String> processorSupplier = kafkaStreamsTracing.processor( "forward-1", () -> new AbstractProcessor<String, String>() { @Override public void process(String key, String value) { assertThat(BAGGAGE_FIELD.getValue(currentTraceContext.get())).isEqualTo("user1"); } }); Headers headers = new RecordHeaders().add(BAGGAGE_FIELD_KEY, "user1".getBytes()); Processor<String, String> processor = processorSupplier.get(); processor.init(processorContextSupplier.apply(headers)); processor.process(TEST_KEY, TEST_VALUE); }