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

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

Use a Decorator pattern or Interceptors to instrument your Java producer and consumer application code for tracing.

.Procedure

In the application code of each 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.

* To use a Decorator pattern:
+
[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);
----

* To use Interceptors:
+
[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. 

To use a Decorator pattern to instrument your producer and consumer applications, 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.

.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()"`.

|===