Functions for integrating Kafka tracing with DataStreams.

Usage

Because Elixir does not include a context grab bag to pass around, we use the OpenTelemetry context to store the current DataStreams pathway. If you are not using OpenTelemetry, or have a special fan in or fan out situation, you can use the respective functions that take a pathway as an argument.

If you have a basic one message in one message out situation, and you have OpenTelemetry already covering your application, you can use the trace_produce/1 and trace_consume/2 functions.

require OpenTelemetry.Tracer, as: Tracer

alias Datadog.DataStreams.Integrations.Kafka, as: DataStreamsKafka

@doc "Handles a message from Kafka. Receives a message map with partition, topic, and headers."
@spec handle_message(map()) :: :ok
def handle_message(message) do
  # NOTE: This does not add the recommended Kafka span attributes.
  Tracer.with_span "#{message.topic} process" do
    DataStreamsKafka.trace_consume(message, "my_consumer_group")

    # Do work

    Tracer.with_span "#{new_message.topic} produce" do
      new_message
      |> DataStreamsKafka.trace_produce()
      |> send_to_kafka()
    end
  end
end