// Module included in the following assemblies:
//
// assembly-distributed tracing.adoc

[id='proc-instrumenting-kafka-streams-with-tracers-{context}']
= Instrumenting Kafka Streams applications for tracing

[role="_abstract"]
Instrument application code to enable tracing in Kafka Streams API applications. 
Use a decorator pattern or interceptors to instrument your Kafka Streams API applications for tracing.
You can then record traces when messages are produced or retrieved from a topic.

Decorator instrumentation:: For decorator instrumentation, create a modified Kafka Streams instance for tracing. The OpenTracing instrumentation project provides a `TracingKafkaClientSupplier` class that supports instrumentation of Kafka Streams.
You create a wrapped instance of the `TracingKafkaClientSupplier` supplier interface, which provides tracing instrumentation for Kafka Streams.
For OpenTelemetry, the process is the same but you need to create a custom `TracingKafkaClientSupplier` class to provide the support. 
Interceptor instrumentation:: For interceptor instrumentation, add the tracing capability to the Kafka Streams producer and consumer configuration. 

.Prerequisites

* You have xref:proc-configuring-tracers-kafka-clients-{context}[initialized tracing for the client].
+
You enable instrumentation in Kafka Streams applications by adding the tracing JARs as dependencies to your project.
* To instrument Kafka Streams with OpenTelemetry, you'll need to write a custom `TracingKafkaClientSupplier`.
* The custom `TracingKafkaClientSupplier` can extend Kafka's `DefaultKafkaClientSupplier`, overriding the producer and consumer creation methods to wrap the instances with the telemetry-related code.
+
.Example custom `TracingKafkaClientSupplier`
[source,java]
----
private class TracingKafkaClientSupplier extends DefaultKafkaClientSupplier {
    @Override
    public Producer<byte[], byte[]> getProducer(Map<String, Object> config) {
        KafkaTelemetry telemetry = KafkaTelemetry.create(GlobalOpenTelemetry.get());
        return telemetry.wrap(super.getProducer(config));
    }

    @Override
    public Consumer<byte[], byte[]> getConsumer(Map<String, Object> config) {
        KafkaTelemetry telemetry = KafkaTelemetry.create(GlobalOpenTelemetry.get());
        return telemetry.wrap(super.getConsumer(config));
    }

    @Override
    public Consumer<byte[], byte[]> getRestoreConsumer(Map<String, Object> config) {
        return this.getConsumer(config);
    }

    @Override
    public Consumer<byte[], byte[]> getGlobalConsumer(Map<String, Object> config) {
        return this.getConsumer(config);
    }
}
----

.Procedure

Perform these steps for each Kafka Streams API application.

* To use a decorator pattern, create an instance of the `TracingKafkaClientSupplier` supplier interface, then provide the supplier interface to `KafkaStreams`.
+
.Example decorator instrumentation
[source,java,subs=attributes+]
----
KafkaClientSupplier supplier = new TracingKafkaClientSupplier(tracer);
KafkaStreams streams = new KafkaStreams(builder.build(), new StreamsConfig(config), supplier);
streams.start();
----

* To use interceptors, set the interceptor class in the Kafka Streams producer and consumer configuration.
+
The `TracingProducerInterceptor` and `TracingConsumerInterceptor` interceptor classes take care of the tracing capability.
+
.Example producer and consumer configuration using interceptors
[source,java,subs=attributes+]
----
props.put(StreamsConfig.PRODUCER_PREFIX + ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, TracingProducerInterceptor.class.getName());
props.put(StreamsConfig.CONSUMER_PREFIX + ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, TracingConsumerInterceptor.class.getName());
----