Chapter 15. Distributed tracing

1. Add Maven dependencies for Jaeger to the pom.xml file for the client application:

   <dependency>
       <groupId>io.jaegertracing</groupId>
       <artifactId>jaeger-client</artifactId>
       <version>1.1.0.redhat-00002</version>
   </dependency>

2. Define the configuration of the Jaeger tracer using the tracing environment variables.

3. Create the Jaeger tracer from the environment variables that you defined in step two:

   Tracer tracer = Configuration.fromEnv().getTracer();

   Note

   For alternative ways to initialize a Jaeger tracer, see the Java OpenTracing library documentation.

4. Register the Jaeger tracer as a global tracer:

   GlobalTracer.register(tracer);

1. Add a Maven dependency for OpenTracing to the producer or consumer’s pom.xml file.

   <dependency>
       <groupId>io.opentracing.contrib</groupId>
       <artifactId>opentracing-kafka-client</artifactId>
       <version>0.1.15.redhat-00001</version>
   </dependency>

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

   • To use a Decorator pattern:

     // 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:

     // 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);

1. Add the opentracing-kafka-streams dependency to the Kafka Streams application’s pom.xml file.

   <dependency>
       <groupId>io.opentracing.contrib</groupId>
       <artifactId>opentracing-kafka-streams</artifactId>
       <version>0.1.15.redhat-00001</version>
   </dependency>

2. Create an instance of the TracingKafkaClientSupplier supplier interface:

   KafkaClientSupplier supplier = new TracingKafkaClientSupplier(tracer);

3. Provide the supplier interface to KafkaStreams:

   KafkaStreams streams = new KafkaStreams(builder.build(), new StreamsConfig(config), supplier);
   streams.start();

Procedure

1. Configure and enable a Jaeger tracer.

2. Edit the /opt/kafka/config/consumer.properties file.

   Add the following Interceptor property:

   consumer.interceptor.classes=io.opentracing.contrib.kafka.TracingConsumerInterceptor

3. Edit the /opt/kafka/config/producer.properties file.

   Add the following Interceptor property:

   producer.interceptor.classes=io.opentracing.contrib.kafka.TracingProducerInterceptor

4. Start MirrorMaker with the consumer and producer configuration files as parameters:

   su - kafka
   /opt/kafka/bin/kafka-mirror-maker.sh --consumer.config /opt/kafka/config/consumer.properties --producer.config /opt/kafka/config/producer.properties --num.streams=2

Procedure

1. Configure and enable a Jaeger tracer.

2. Edit the MirrorMaker 2.0 configuration properties file, ./config/connect-mirror-maker.properties, and add the following properties:

   header.converter=org.apache.kafka.connect.converters.ByteArrayConverter 1
   consumer.interceptor.classes=io.opentracing.contrib.kafka.TracingConsumerInterceptor 2
   producer.interceptor.classes=io.opentracing.contrib.kafka.TracingProducerInterceptor

   1

   Prevents Kafka Connect from converting message headers (containing trace IDs) to base64 encoding. This ensures that messages are the same in both the source and the target clusters.

   2

   Enables the Interceptors for MirrorMaker 2.0.

3. Start MirrorMaker 2.0 using the instructions in Section 9.4, “Synchronizing data between Kafka clusters using MirrorMaker 2.0”.

Procedure

1. Configure and enable a Jaeger tracer.

2. Edit the relevant Kafka Connect configuration file.

   • If you are running Kafka Connect in standalone mode, edit the /opt/kafka/config/connect-standalone.properties file.
   • If you are running Kafka Connect in distributed mode, edit the /opt/kafka/config/connect-distributed.properties file.

3. Add the following properties to the configuration file:

   producer.interceptor.classes=io.opentracing.contrib.kafka.TracingProducerInterceptor
   consumer.interceptor.classes=io.opentracing.contrib.kafka.TracingConsumerInterceptor

4. Save the configuration file.
5. Set tracing environment variables and then run Kafka Connect in standalone or distributed mode.