# karafka.rb
class KarafkaApp < Karafka::App
  setup do |config|
    config.kafka = { 'bootstrap.servers': '127.0.0.1:9092' }
  end

  routes.draw do
    topic :example do
      consumer ExampleConsumer
    end
  end
end

# app/consumers/example_consumer.rb
class ExampleConsumer < ApplicationConsumer
  def consume
    log("before #each")

    messages.each do |message|
      log("inside #each")
      puts "doing some hard work!"
    end

    log("after #each")
  end

  def log(description)
    trace = Datadog::Tracing.active_trace
    span = Datadog::Tracing.active_span

    data = {id: trace&.id, name: trace&.name, resource: trace&.resource, parent_span_id: trace&.parent_span_id}
    puts "[consumer - #{description}] trace: #{data}"

    data = {id: span&.id, name: span&.name, resource: span&.resource, parent_id: span&.parent_id}
    puts "[consumer - #{description}] span: #{data}"
  end
end

# producing messages in a terminal
["msg1", "msg2"].each do |msg|
  Datadog::Tracing.trace("producer.#{msg}") do
    digest = Datadog::Tracing.active_trace.to_digest
    headers = {}
    Datadog::Tracing::Contrib::Karafka.inject(digest, headers)
    Karafka.producer.produce_async({
      topic: "example", payload: { foo: msg }.to_json, headers: headers
    })
  end
end

Datadog.configure do |c|
  c.tracing.instrument :karafka, distributed_tracing: true, enabled: true
end