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

[id='proc-instrumenting-producers-consumers-for-opentracing-{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 and OpenTracing instrumentation projects provide classes that support instrumentation of producers and consumers.

Decorator instrumentation:: For decorator instrumentation, create a modified producer or consumer instance for tracing. Decorator instrumentation is different for OpenTelemetry and OpenTracing.
Interceptor instrumentation:: For interceptor instrumentation, add the tracing capability to the consumer or producer configuration. Interceptor instrumentation is the same for OpenTelemetry and OpenTracing.

.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);     
----
+
.Example decorator instrumentation for OpenTracing
[source,java]
----
//producer instance
KafkaProducer<Integer, String> producer = new KafkaProducer<>(senderProps);
TracingKafkaProducer<Integer, String> tracingProducer = new TracingKafkaProducer<>(producer, tracer);
TracingKafkaProducer.send(...)

//consumer instance
KafkaConsumer<Integer, String> consumer = new KafkaConsumer<>(consumerProps);
TracingKafkaConsumer<Integer, String> tracingConsumer = new TracingKafkaConsumer<>(consumer, tracer);
tracingConsumer.subscribe(Collections.singletonList("mytopic"));
ConsumerRecords<Integer, String> records = tracingConsumer.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);
----