9.3. 使用 tracer 检查 Kafka 客户端

检测 Kafka producer 和使用者客户端，以及用于分布式追踪的 Kafka Streams API 应用程序。

9.3.1. 提取制作者和消费者以进行追踪
复制链接

使用 Decorator 模式或 Interceptors 来检测您的 Java 制作者和消费者应用程序代码进行追踪。

流程

在每个制作者和消费者应用程序的应用程序代码中：

将 OpenTracing 的 Maven 依赖项添加到制作者或消费者的 pom.xml 文件。

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

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

Copy to Clipboard

Toggle word wrap

使用 Decorator 模式或 Interceptors 检测您的客户端应用程序代码。

使用 Decorator 模式：

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

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

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

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

9.3.1.1. 在 Decorator 模式中的自定义 span 名称
复制链接

span 是 Jaeger 中的逻辑工作单元，包括操作名称、开始时间和持续时间。

要使用 Decorator 模式检测您的制作者和消费者应用程序，请在创建 TracingKafkaProducer 和 TracingKafkaConsumer 对象时将 BiFunction 对象作为附加参数来定义自定义 span 名称。OpenTracing Apache Kafka Client Instrumentation 库包括几个内置范围名称。

示例：使用自定义 span 名称以 Decorator 模式检测客户端应用程序代码

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

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

Copy to Clipboard

Toggle word wrap

9.3.1.2. 内置范围名称
复制链接

在定义自定义范围名称时，您可以在 ClientSpanNameProvider 类中使用以下 BiFunctions。如果没有指定 spanNameProvider，则使用 CONSUMER_OPERATION_NAME 和 PRODUCER_OPERATION_NAME。

Expand

BiFunction	Description
`CONSUMER_OPERATION_NAME, PRODUCER_OPERATION_NAME`	返回 `operationName` 作为使用者和"end"的 span name: "receive"。
`CONSUMER_PREFIXED_OPERATION_NAME （字符串前缀）、PRODUCER_PREFIXED_OPERATION_NAME （字符串前缀）`	返回 `前缀` 和 `operationName` 的字符串。
`CONSUMER_TOPIC, PRODUCER_TOPIC`	以格式 `(record.topic （)）` 返回消息发送到或检索的主题名称。
`PREFIXED_CONSUMER_TOPIC （字符串前缀）、PREFIXED_PRODUCER_TOPIC （字符串前缀）`	返回 `前缀的` String connection，以及格式为 `(record.topic （)）` 的主题名称。
`CONSUMER_OPERATION_NAME_TOPIC, PRODUCER_OPERATION_NAME_TOPIC`	返回操作名称和主题名称：" `operationName - record.topic （）"`。
`CONSUMER_PREFIXED_OPERATION_NAME_TOPIC （字符串前缀）、PRODUCER_PREFIXED_OPERATION_NAME_TOPIC （字符串前缀）`	返回 `前缀` String linkion 和 `"operationName - record.topic （）"`。

9.3.2. 用于追踪的 Kafka Streams 应用程序
复制链接

本节论述了如何使用 Kafka Streams API 应用程序进行分布式追踪。

流程

在每个 Kafka Streams API 应用程序中：

将 opentracing-kafka-streams 依赖项添加到 Kafka Streams API 应用程序的 pom.xml 文件中：

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

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

Copy to Clipboard

Toggle word wrap

创建 TracingKafkaClientSupplier 供应商接口实例：

KafkaClientSupplier supplier = new TracingKafkaClientSupplier(tracer);

KafkaClientSupplier supplier = new TracingKafkaClientSupplier(tracer);

Copy to Clipboard

Toggle word wrap

为 KafkaStreams 提供供应商接口：

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

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

Copy to Clipboard

Toggle word wrap

返回顶部

9.3. 使用 tracer 检查 Kafka 客户端

9.3.1. 提取制作者和消费者以进行追踪
复制链接

9.3.1.1. 在 Decorator 模式中的自定义 span 名称
复制链接

9.3.1.2. 内置范围名称
复制链接

9.3.2. 用于追踪的 Kafka Streams 应用程序
复制链接

学习

尝试、购买和销售

社区

关于红帽文档

让开源更具包容性

關於紅帽

Theme

Red Hat legal and privacy links

Red Hat legal and privacy links

9.3. 使用 tracer 检查 Kafka 客户端

9.3.1. 提取制作者和消费者以进行追踪复制链接链接已复制到粘贴板!

9.3.1.1. 在 Decorator 模式中的自定义 span 名称复制链接链接已复制到粘贴板!

9.3.1.2. 内置范围名称复制链接链接已复制到粘贴板!

9.3.2. 用于追踪的 Kafka Streams 应用程序复制链接链接已复制到粘贴板!

学习

尝试、购买和销售

社区

关于红帽文档

让开源更具包容性

關於紅帽

Theme

Red Hat legal and privacy links

Red Hat legal and privacy links

9.3.1. 提取制作者和消费者以进行追踪
复制链接

9.3.1.1. 在 Decorator 模式中的自定义 span 名称
复制链接

9.3.1.2. 内置范围名称
复制链接

9.3.2. 用于追踪的 Kafka Streams 应用程序
复制链接