KafkaStreamsTracing { Span nextSpan(ProcessorContext context) { TraceContextOrSamplingFlags extracted = extractor.extract(context.headers()); if (!extracted.equals(emptyExtraction)) { clearHeaders(context.headers()); } Span result = tracer.nextSpan(extracted); if (!result.isNoop()) { addTags(context, result); } return result; } 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 nextSpan_uses_current_context() { ProcessorContext fakeProcessorContext = processorContextSupplier.apply(new RecordHeaders()); Span child; try (Scope ws = tracing.currentTraceContext().newScope(parent)) { child = kafkaStreamsTracing.nextSpan(fakeProcessorContext); } child.finish(); assertThat(child.context().parentIdString()) .isEqualTo(parent.spanIdString()); }
@Test public void nextSpan_should_create_span_if_no_headers() { ProcessorContext fakeProcessorContext = processorContextSupplier.apply(new RecordHeaders()); assertThat(kafkaStreamsTracing.nextSpan(fakeProcessorContext)).isNotNull(); }
@Test public void nextSpan_should_tag_app_id_and_task_id() { ProcessorContext fakeProcessorContext = processorContextSupplier.apply(new RecordHeaders()); kafkaStreamsTracing.nextSpan(fakeProcessorContext).start().finish(); assertThat(spans.get(0).tags()) .containsOnly( entry("kafka.streams.application.id", TEST_APPLICATION_ID), entry("kafka.streams.task.id", TEST_TASK_ID)); }