DdDataStreams.Extractor.Kafka (DD Data Streams v0.3.0)

Copy Markdown View Source

The extractor for BroadwayKafka 0.7.1.

A Kafka record names its consume edge by its topic, which BroadwayKafka stamps as metadata.topic together with metadata.headers, metadata.key and metadata.ts. A message that carries no topic records no consume checkpoint, because there is no name to report its edge under. It was checked against the source of BroadwayKafka 0.7.1 and brod 4.7.0, not against a live broker.

Both edges are reported as type:kafka:

  • consume - direction:in, topic:<topic>, and group:<consumer group> when the producer's :group_id option is known
  • produce - direction:out and topic:<topic>

The consumer group is read from the producer Broadway reports for the message, so it needs a core that hands the producer to c:consume/2; without one the group: tag is left out. partition and offset are not reported, as the Datadog tracers do not report them on a Data Streams edge either: a per-partition tag would name whichever partition arrived first for the node. The kafka_cluster_id tag the tracers add is not reported, because BroadwayKafka exposes no cluster id and rejects producer options it does not know; a node of this library therefore differs from a tracer node that reports one, though the pathway still connects through the header.

The record headers are the carrier, so the pathway header round-trips with DdDataStreams.Carrier.Flat. A record without headers (nil or :undefined) is read as one with no inbound pathway. The Kafka key is the ordering key, so it counts towards the payload size as it does in the tracers.

metadata.ts is a publish time, in Unix milliseconds: the producer's CreateTime, or the broker's LogAppendTime on a topic configured that way, and BroadwayKafka does not say which. It is offered as :pathway_start and :edge_start, converted to native time units, and never as :received_at, which stays the time the message reached the pipeline. A first-hop message therefore measures from it, as dd-trace-java does (dd-trace-go and dd-trace-py stamp the consume time); one that carries a pathway header keeps the start its producer wrote, and a consumer replaying a backlog reports the record's age. A timestamp that is :undefined (a magic-0 record) or not positive is dropped, as is one the core finds outside its window.