Chapter 16. 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>

   Copy to Clipboard Toggle word wrap
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();

   Copy to Clipboard Toggle word wrap
   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);

   Copy to Clipboard Toggle word wrap

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.12.redhat-00001</version>
   </dependency>

   Copy to Clipboard Toggle word wrap

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

     Copy to Clipboard Toggle word wrap

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

     Copy to Clipboard Toggle word wrap

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.12.redhat-00001</version>
   </dependency>

   Copy to Clipboard Toggle word wrap

2. Create an instance of the TracingKafkaClientSupplier supplier interface:

   KafkaClientSupplier supplier = new TracingKafkaClientSupplier(tracer);

   Copy to Clipboard Toggle word wrap

3. Provide the supplier interface to KafkaStreams:

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

   Copy to Clipboard Toggle word wrap

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

   Copy to Clipboard Toggle word wrap

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

   Add the following Interceptor property:

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

   Copy to Clipboard Toggle word wrap

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

   Copy to Clipboard Toggle word wrap

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

   Copy to Clipboard Toggle word wrap

   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 10.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

   Copy to Clipboard Toggle word wrap
4. Save the configuration file.
5. Set tracing environment variables and then run Kafka Connect in standalone or distributed mode.