Module: Datadog::Tracing::Contrib::Kafka::Instrumentation::Consumer

Defined in:
lib/datadog/tracing/contrib/kafka/instrumentation/consumer.rb

Overview

Instrumentation for Kafka::Consumer

Instance Method Summary collapse

Instance Method Details

#each_batch(**kwargs, &block) ⇒ Object 



28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
 
# File 'lib/datadog/tracing/contrib/kafka/instrumentation/consumer.rb', line 28

def each_batch(**kwargs, &block)
  if Datadog::DataStreams.enabled?
    super do |batch|
      Datadog::DataStreams.set_consume_checkpoint(
        type: 'kafka',
        source: batch.topic,
        auto_instrumentation: true
      )

      block.call(batch)
    end
  else
    super
  end
end

#each_message(**kwargs, &block) ⇒ Object 



10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
 
# File 'lib/datadog/tracing/contrib/kafka/instrumentation/consumer.rb', line 10

def each_message(**kwargs, &block)
  if Datadog::DataStreams.enabled?
    super do |message|
      headers = message.headers || {}

      Datadog::DataStreams.set_consume_checkpoint(
        type: 'kafka',
        source: message.topic,
        auto_instrumentation: true
      ) { |key| headers[key] }

      block.call(message)
    end
  else
    super
  end
end