// Module included in the following assemblies:
//
// assembly-instrumenting-kafka-clients-tracers.adoc

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

Use a Decorator pattern or Interceptors to instrument your Java Producer and Consumer application code for distributed tracing.

.Procedure

Perform these steps in the application code of each Kafka Producer and Consumer application.

. Add the Maven dependency for OpenTracing to the Producer or Consumer's `pom.xml` file.
+
[source,xml,subs="attributes+"]
----
<dependency>
    <groupId>io.opentracing.contrib</groupId>
    <artifactId>opentracing-kafka-client</artifactId>
    <version>{OpenTracingKafkaClient}</version>
</dependency>
----

. Instrument your client application code using either a Decorator pattern or Interceptors.

* If you prefer to use a Decorator pattern, use following example:
+
[source,java,subs=attributes+]
----
// Create an instance of the KafkaProducer:
KafkaProducer<Integer, String> producer = new KafkaProducer<>(senderProps);

// Create an instance of the TracingKafkaProducer:
TracingKafkaProducer<Integer, String> tracingProducer = new TracingKafkaProducer<>(producer, 
        tracer);

// Send:
tracingProducer.send(...);

// Create an instance of the KafkaConsumer:
KafkaConsumer<Integer, String> consumer = new KafkaConsumer<>(consumerProps);

// Create an instance of the TracingKafkaConsumer:
TracingKafkaConsumer<Integer, String> tracingConsumer = new TracingKafkaConsumer<>(consumer, 
        tracer);

// Subscribe:
tracingConsumer.subscribe(Collections.singletonList("messages"));

// Get messages:
ConsumerRecords<Integer, String> records = tracingConsumer.poll(1000);

// Retrieve SpanContext from polled record (consumer side):
ConsumerRecord<Integer, String> record = ...
SpanContext spanContext = TracingKafkaUtils.extractSpanContext(record.headers(), tracer);
----

* If you prefer to use Interceptors, use the following example:
+
[source,java,subs=attributes+]
----
// Register the tracer with GlobalTracer:
GlobalTracer.register(tracer);

// Add the TracingProducerInterceptor to the sender properties:
senderProps.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, 
          TracingProducerInterceptor.class.getName());

// Create an instance of the KafkaProducer:
KafkaProducer<Integer, String> producer = new KafkaProducer<>(senderProps);

// Send:
producer.send(...);

// Add the TracingConsumerInterceptor to the consumer properties:
consumerProps.put(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG,
          TracingConsumerInterceptor.class.getName());

// Create an instance of the KafkaConsumer:
KafkaConsumer<Integer, String> consumer = new KafkaConsumer<>(consumerProps);

// Subscribe:
consumer.subscribe(Collections.singletonList("messages"));

// Get messages:
ConsumerRecords<Integer, String> records = consumer.poll(1000);

// Retrieve the SpanContext from a polled message (consumer side):
ConsumerRecord<Integer, String> record = ...
SpanContext spanContext = TracingKafkaUtils.extractSpanContext(record.headers(), tracer);

----

== Custom span names in a Decorator pattern

A __span__ is a logical unit of work in Jaeger, with an operation name, start time, and duration. 

If you use a Decorator pattern to instrument your Kafka Producer and Consumer applications, you can define custom span names by passing a `BiFunction` object as an additional argument when creating the `TracingKafkaProducer` and `TracingKafkaConsumer` objects. The OpenTracing Apache Kafka Client Instrumentation library includes several built-in span names, which are described below.

.Example: Using custom span names to instrument client application code in a Decorator pattern

[source,java,subs=attributes+]
----
// Create a BiFunction for the KafkaProducer that operates on (String operationName, ProducerRecord consumerRecord) and returns a String to be used as the name:

BiFunction<String, ProducerRecord, String> producerSpanNameProvider =
    (operationName, producerRecord) -> "CUSTOM_PRODUCER_NAME";

// Create an instance of the KafkaProducer:
KafkaProducer<Integer, String> producer = new KafkaProducer<>(senderProps);

// Create an instance of the TracingKafkaProducer
TracingKafkaProducer<Integer, String> tracingProducer = new TracingKafkaProducer<>(producer, 
        tracer,
        producerSpanNameProvider);

// Spans created by the tracingProducer will now have "CUSTOM_PRODUCER_NAME" as the span name.

// Create a BiFunction for the KafkaConsumer that operates on (String operationName, ConsumerRecord consumerRecord) and returns a String to be used as the name:

BiFunction<String, ConsumerRecord, String> consumerSpanNameProvider =
    (operationName, consumerRecord) -> operationName.toUpperCase();

// Create an instance of the KafkaConsumer:
KafkaConsumer<Integer, String> consumer = new KafkaConsumer<>(consumerProps);

// Create an instance of the TracingKafkaConsumer, passing in the consumerSpanNameProvider BiFunction:

TracingKafkaConsumer<Integer, String> tracingConsumer = new TracingKafkaConsumer<>(consumer, 
        tracer,
        consumerSpanNameProvider);

// Spans created by the tracingConsumer will have the operation name as the span name, in upper-case.
// "receive" -> "RECEIVE"
----

== Built-in span names

When defining custom span names, you can use the following `BiFunctions` in the `ClientSpanNameProvider` class. If no `spanNameProvider` is specified, `CONSUMER_OPERATION_NAME` and `PRODUCER_OPERATION_NAME` are used.

[%autowidth,cols="2*",options="header",stripes="none",separator=¦]
|===

¦BiFunction
¦Description

m¦CONSUMER_OPERATION_NAME, PRODUCER_OPERATION_NAME
¦Returns the `operationName` as the span name: "receive" for Consumers and "send" for Producers.

m¦CONSUMER_PREFIXED_OPERATION_NAME(String prefix), PRODUCER_PREFIXED_OPERATION_NAME(String prefix)
¦Returns a String concatenation of `prefix` and `operationName`.

m¦CONSUMER_TOPIC, PRODUCER_TOPIC
¦Returns the name of the topic that the message was sent to or retrieved from in the format `(record.topic())`.

m¦PREFIXED_CONSUMER_TOPIC(String prefix), PREFIXED_PRODUCER_TOPIC(String prefix)
¦Returns a String concatenation of `prefix` and the topic name in the format `(record.topic())`.

m¦CONSUMER_OPERATION_NAME_TOPIC, PRODUCER_OPERATION_NAME_TOPIC
¦Returns the operation name and the topic name: `"operationName - record.topic()"`.

m¦CONSUMER_PREFIXED_OPERATION_NAME_TOPIC(String prefix), PRODUCER_PREFIXED_OPERATION_NAME_TOPIC(String prefix)
¦Returns a String concatenation of `prefix` and `"operationName - record.topic()"`.

|===