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

[id='proc-instrumenting-producers-consumers-for-tracing-{context}']
= Instrumenting producers and consumers for tracing

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

OpenTelemetry instrumentation project provides classes that support instrumentation of producers and consumers.

Decorator instrumentation:: For decorator instrumentation, create a modified producer or consumer instance for tracing.
Interceptor instrumentation:: For interceptor instrumentation, add the tracing capability to the consumer or producer configuration.

.Prerequisites

* You have xref:proc-configuring-tracers-kafka-clients-{context}[initialized tracing for the client].
+
You enable instrumentation in producer and consumer applications by adding the tracing JARs as dependencies to your project.

.Procedure

Perform these steps in the application code of each producer and consumer application.
Instrument your client application code using either a decorator pattern or interceptors.

* To use a decorator pattern, create a modified producer or consumer instance to send or receive messages. 
+
You pass the original `KafkaProducer` or `KafkaConsumer` class.
+
.Example decorator instrumentation for OpenTelemetry
[source,java]
----
// Producer instance
Producer < String, String > op = new KafkaProducer < > (
    configs,
    new StringSerializer(),
    new StringSerializer()
    );
    Producer < String, String > producer = tracing.wrap(op);
KafkaTracing tracing = KafkaTracing.create(GlobalOpenTelemetry.get());
producer.send(...);

//consumer instance
Consumer<String, String> oc = new KafkaConsumer<>(
    configs,
    new StringDeserializer(),
    new StringDeserializer()
    );
    Consumer<String, String> consumer = tracing.wrap(oc);
consumer.subscribe(Collections.singleton("mytopic"));
ConsumerRecords<Integer, String> records = consumer.poll(1000);
ConsumerRecord<Integer, String> record = ...
SpanContext spanContext = TracingKafkaUtils.extractSpanContext(record.headers(), tracer);     
----

* To use interceptors, set the interceptor class in the producer or consumer configuration.
+
You use the `KafkaProducer` and `KafkaConsumer` classes in the usual way.
The `TracingProducerInterceptor` and `TracingConsumerInterceptor` interceptor classes take care of the tracing capability.
+
.Example producer configuration using interceptors
[source,java]
----
senderProps.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, 
    TracingProducerInterceptor.class.getName());

KafkaProducer<Integer, String> producer = new KafkaProducer<>(senderProps);
producer.send(...);
----
+
.Example consumer configuration using interceptors
[source,java]
----
consumerProps.put(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG,
    TracingConsumerInterceptor.class.getName());

KafkaConsumer<Integer, String> consumer = new KafkaConsumer<>(consumerProps);
consumer.subscribe(Collections.singletonList("messages"));
ConsumerRecords<Integer, String> records = consumer.poll(1000);
ConsumerRecord<Integer, String> record = ...
SpanContext spanContext = TracingKafkaUtils.extractSpanContext(record.headers(), tracer);
----