16.8. プロデューサーおよびコンシューマーをトレース用にインストルメント化

前提条件

• クライアントのトレースを初期化 している。

  トレース JAR を依存関係としてプロジェクトに追加して、プロデューサーアプリケーションとコンシューマーアプリケーションでインストルメント化を有効にしている。

• デコレーターパターンを使用するには、変更したプロデューサーまたはコンシューマーインスタンスを作成して、メッセージを送受信します。

  元の KafkaProducer または KafkaConsumer クラスを渡します。

  OpenTelemetry のデコレーターインストルメント化の例

  // Producer instance
  Producer < String, String > op = new KafkaProducer < > (
      configs,
      new StringSerializer(),
      new StringSerializer()
      );
      Producer < String, String > producer = tracing.wrap(op);
  KafkaTracing tracing = KafkaTracing.create(GlobalOpenTelemetry.get());
  producer.send(...);
  
  //consumer instance
  Consumer<String, String> oc = new KafkaConsumer<>(
      configs,
      new StringDeserializer(),
      new StringDeserializer()
      );
      Consumer<String, String> consumer = tracing.wrap(oc);
  consumer.subscribe(Collections.singleton("mytopic"));
  ConsumerRecords<Integer, String> records = consumer.poll(1000);
  ConsumerRecord<Integer, String> record = ...
  SpanContext spanContext = TracingKafkaUtils.extractSpanContext(record.headers(), tracer);

• インターセプターを使用するには、プロデューサーまたはコンシューマーの設定でインターセプタークラスを設定します。

  通常の方法で KafkaProducer クラスと KafkaConsumer クラスを使用します。TracingProducerInterceptor および TracingConsumerInterceptor インターセプタークラスは、トレース機能を処理します。

  インターセプターを使用したプロデューサー設定の例

  senderProps.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,
      TracingProducerInterceptor.class.getName());
  
  KafkaProducer<Integer, String> producer = new KafkaProducer<>(senderProps);
  producer.send(...);

  インターセプターを使用したコンシューマー設定の例

  consumerProps.put(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG,
      TracingConsumerInterceptor.class.getName());
  
  KafkaConsumer<Integer, String> consumer = new KafkaConsumer<>(consumerProps);
  consumer.subscribe(Collections.singletonList("messages"));
  ConsumerRecords<Integer, String> records = consumer.poll(1000);
  ConsumerRecord<Integer, String> record = ...
  SpanContext spanContext = TracingKafkaUtils.extractSpanContext(record.headers(), tracer);