feat(kafkajs): add kafka.messages.offsets to producer span#8468
Merged
Conversation
Mirror dd-trace-java PR #11107 for kafkajs: set kafka.partition and kafka.message.offset on the producer span using the broker-returned RecordMetadata. Only tagged when the send resolves to a single (topic, partition) tuple, since one producer span covers a multi-partition batch and a single value would be misleading. kafka.cluster_id was already tagged on both producer and consumer spans, and the consumer span already carried partition and offset, so no additional changes are needed for those tags. Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
Contributor
Overall package sizeSelf size: 5.85 MB Dependency sizes| name | version | self size | total size | |------|---------|-----------|------------| | import-in-the-middle | 3.0.1 | 82.56 kB | 817.39 kB | | dc-polyfill | 0.1.11 | 25.74 kB | 25.74 kB |🤖 This report was automatically generated by heaviest-objects-in-the-universe |
🎉 All green!❄️ No new flaky tests detected 🎯 Code Coverage (details) 🔗 Commit SHA: 1f113ca | Docs | Datadog PR Page | Give us feedback! |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: e88a20ad89
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
| span.setTag('kafka.partition', partition) | ||
| } | ||
| if (offsetAsLong !== undefined) { | ||
| span.setTag('kafka.message.offset', Number(offsetAsLong)) |
There was a problem hiding this comment.
Avoid coercing the broker offset to Number here: Kafka offsets are 64-bit values, so once a topic offset grows past Number.MAX_SAFE_INTEGER this conversion silently loses precision. In those high-throughput/long-lived topics, kafka.message.offset on producer spans no longer matches the exact message offset, which undermines the commit’s stated goal of linking producer and consumer spans by (cluster, topic, partition, offset).
Useful? React with 👍 / 👎.
…messages.offsets Replace single-value kafka.partition and kafka.message.offset on the producer span with a single kafka.messages.offsets tag carrying the full list of (partition, start_offset) pairs returned by the broker. One producer span covers a multi-partition batch, so a list captures every partition the batch landed in rather than only the single-partition case. Total record count is still available via the pre-existing kafka.batch_size metric. Per-partition record counts are not in the broker response and would require replicating kafkajs's partitioner, so they are not included. Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
piochelepiotr
changed the title
Restore kafka.partition and kafka.message.offset flat tags on the producer span, but only when the send is a single-record batch. In that case the one response entry's partition/offset describes the exact record, so the flat tags are unambiguous. Multi-message batches still expose data through kafka.messages.offsets only. Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
Mirror the producer change on the batch consumer side. kafkajs's eachBatch hands the user one batch per (topic, partition), so all records share a single partition and a known start offset (the first record's offset). Tag kafka.messages.offsets accordingly, and surface the flat kafka.message.offset metric only when the batch is a single record — matching the producer's single-record convention. Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
BenchmarksBenchmark execution time: 2026-05-14 15:07:46 Comparing candidate commit 1f113ca in PR branch Found 0 performance improvements and 0 performance regressions! Performance is the same for 1508 metrics, 85 unstable metrics. |
…er span" eachBatch is always single-partition by construction, so the list-form tag is redundant with the flat kafka.partition that's already set on the consume-batch span. Revert and leave the consume-batch span as it was on master. Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
chnn
previously approved these changes
May 13, 2026
crysmags
requested changes
May 14, 2026
Collaborator
There was a problem hiding this comment.
Two blockers, a few nits and otherwise this looks good.
offsets are 64-bit, Number(longValue) rounds past 2^53. on busy topics this breaks the correlation the PR is trying to make, codex flagged the same thing above
the multi-message test still produces a single-entry list because the topic is single-partition — so the headline case isn't covered.
| for (const entry of result) { | ||
| const offsetAsLong = entry.offset || entry.baseOffset | ||
| if (entry.partition === undefined || offsetAsLong === undefined) continue | ||
| offsets.push({ partition: entry.partition, start_offset: Number(offsetAsLong) }) |
Collaborator
There was a problem hiding this comment.
this loses precision on long-lived topics. keep it a string:
offsets.push({ partition: entry.partition, start_offset: String(offsetAsLong) })flat kafka.message.offset has the same issue but it's a metric in the test, so it has to move to meta to be safe. fine as a follow-up if you'd rather keep this PR small.
Collaborator
There was a problem hiding this comment.
Also, can you add a regression test that exercises a > 2^53 offset? commit.spec.js uses sinon and you could too to mock this.
| // response, only the starting offset. | ||
| const offsets = [] | ||
| for (const entry of result) { | ||
| const offsetAsLong = entry.offset || entry.baseOffset |
Collaborator
There was a problem hiding this comment.
?? instead of || here — defensive against a literal 0.
| offsets.push({ partition: entry.partition, start_offset: Number(offsetAsLong) }) | ||
| } | ||
| if (offsets.length > 0) { | ||
| span.setTag('kafka.messages.offsets', JSON.stringify(offsets)) |
Collaborator
There was a problem hiding this comment.
sort by partition before stringify so the tag is deterministic:
offsets.sort((a, b) => a.partition - b.partition)| assertObjectContains(span, { | ||
| name: expectedSchema.send.opName, | ||
| meta: { | ||
| 'kafka.messages.offsets': JSON.stringify([{ partition: 0, start_offset: 0 }]), |
Collaborator
There was a problem hiding this comment.
this test still asserts a one-entry list:
'kafka.messages.offsets': JSON.stringify([{ partition: 0, start_offset: 0 }])multi-partition isn't actually covered. either bump the topic to numPartitions: 3 or stub ctx.result directly.
Keep offsets as strings to avoid Number() precision loss past 2^53, sort by partition for deterministic tag output, and use ?? so a literal-zero baseOffset isn't dropped. Move kafka.message.offset from metric to meta to match the string type. Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
Merged
BridgeAR
added a commit
that referenced
this pull request
May 19, 2026
This adds tracing, header injection, and DSM checkpoints for the kafkajs producer.sendBatch API, which was previously uninstrumented: calls produced no spans and propagated no trace or DSM context. One kafka.produce span per topicMessages[] entry matches what producer.send already emits for the single-topic case and mirrors how kafkajs implements send internally (producer/messageProducer.js delegates send to a single-entry sendBatch). Per-entry granularity keeps DSM checkpoints correct on multi-topic batches. The broker response is one aggregated array covering every topic, so the plugin's finish filters by ctx.topic before tagging kafka.messages.offsets; otherwise each per-topic span would carry every other topic's offsets. Fixes: #1711 Refs: #8468
BridgeAR
added a commit
that referenced
this pull request
May 19, 2026
This adds tracing, header injection, and DSM checkpoints for the kafkajs producer.sendBatch API, which was previously uninstrumented: calls produced no spans and propagated no trace or DSM context. One kafka.produce span per topicMessages[] entry matches what producer.send already emits for the single-topic case and mirrors how kafkajs implements send internally (producer/messageProducer.js delegates send to a single-entry sendBatch). Per-entry granularity keeps DSM checkpoints correct on multi-topic batches. The broker response is one aggregated array covering every topic, so the plugin's finish filters by ctx.topic before tagging kafka.messages.offsets; otherwise each per-topic span would carry every other topic's offsets. Fixes: #1711 Refs: #8468
pabloerhard
pushed a commit
that referenced
this pull request
May 20, 2026
* feat(kafkajs): tag producer span with partition and offset Mirror dd-trace-java PR #11107 for kafkajs: set kafka.partition and kafka.message.offset on the producer span using the broker-returned RecordMetadata. Only tagged when the send resolves to a single (topic, partition) tuple, since one producer span covers a multi-partition batch and a single value would be misleading. kafka.cluster_id was already tagged on both producer and consumer spans, and the consumer span already carried partition and offset, so no additional changes are needed for those tags. Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]> * refactor(kafkajs): replace producer partition/offset tags with kafka.messages.offsets Replace single-value kafka.partition and kafka.message.offset on the producer span with a single kafka.messages.offsets tag carrying the full list of (partition, start_offset) pairs returned by the broker. One producer span covers a multi-partition batch, so a list captures every partition the batch landed in rather than only the single-partition case. Total record count is still available via the pre-existing kafka.batch_size metric. Per-partition record counts are not in the broker response and would require replicating kafkajs's partitioner, so they are not included. Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]> * test(kafkajs): cover single- and multi-message producer batches Restore kafka.partition and kafka.message.offset flat tags on the producer span, but only when the send is a single-record batch. In that case the one response entry's partition/offset describes the exact record, so the flat tags are unambiguous. Multi-message batches still expose data through kafka.messages.offsets only. Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]> * feat(kafkajs): add kafka.messages.offsets to eachBatch consumer span Mirror the producer change on the batch consumer side. kafkajs's eachBatch hands the user one batch per (topic, partition), so all records share a single partition and a known start offset (the first record's offset). Tag kafka.messages.offsets accordingly, and surface the flat kafka.message.offset metric only when the batch is a single record — matching the producer's single-record convention. Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]> * fix(kafkajs): flip negated condition to satisfy lint Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]> * Revert "feat(kafkajs): add kafka.messages.offsets to eachBatch consumer span" eachBatch is always single-partition by construction, so the list-form tag is redundant with the flat kafka.partition that's already set on the consume-batch span. Revert and leave the consume-batch span as it was on master. Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]> * fix(kafkajs): preserve 64-bit precision in producer offset tags Keep offsets as strings to avoid Number() precision loss past 2^53, sort by partition for deterministic tag output, and use ?? so a literal-zero baseOffset isn't dropped. Move kafka.message.offset from metric to meta to match the string type. Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]> --------- Co-authored-by: Claude Opus 4.7 (1M context) <[email protected]>
BridgeAR
added a commit
that referenced
this pull request
May 22, 2026
This adds tracing, header injection, and DSM checkpoints for the kafkajs producer.sendBatch API, which was previously uninstrumented: calls produced no spans and propagated no trace or DSM context. One kafka.produce span per topicMessages[] entry matches what producer.send already emits for the single-topic case and mirrors how kafkajs implements send internally (producer/messageProducer.js delegates send to a single-entry sendBatch). Per-entry granularity keeps DSM checkpoints correct on multi-topic batches. The broker response is one aggregated array covering every topic, so the plugin's finish filters by ctx.topic before tagging kafka.messages.offsets; otherwise each per-topic span would carry every other topic's offsets. Fixes: #1711 Refs: #8468
BridgeAR
added a commit
that referenced
this pull request
May 26, 2026
* feat(kafkajs): instrument producer.sendBatch This adds tracing, header injection, and DSM checkpoints for the kafkajs producer.sendBatch API, which was previously uninstrumented: calls produced no spans and propagated no trace or DSM context. One kafka.produce span per topicMessages[] entry matches what producer.send already emits for the single-topic case and mirrors how kafkajs implements send internally (producer/messageProducer.js delegates send to a single-entry sendBatch). Per-entry granularity keeps DSM checkpoints correct on multi-topic batches. The broker response is one aggregated array covering every topic, so the plugin's finish filters by ctx.topic before tagging kafka.messages.offsets; otherwise each per-topic span would carry every other topic's offsets. Fixes: #1711 Refs: #8468 * test(kafkajs): cover producer-boundary slow path and disable arms The integration spec in `packages/datadog-plugin-kafkajs/test/index.spec.js` needs Docker Kafka and always connects the producer before sending, so the slow-path branch through `cluster.refreshMetadataIfNecessary().then(...)` and its `refreshMetadataIfNecessary`-missing fallback never ran outside that one CI flag. This adds an instrumentation-level spec that drives the boundary with a faked cluster, pinning the resolve / reject / missing arms of the slow path, the proactive and reactive `refreshHeaderSupport` disable, and the `producerStartCh.hasSubscribers` fast skip without spinning a broker. Refs: #8390 * ci(kafkajs): run the datadog-instrumentations spec in CI The instrumentation-level spec at `packages/datadog-instrumentations/test/kafkajs.spec.js` is matched by `test:instrumentations`'s `PLUGINS=kafkajs` glob, but no workflow runs it. Without a job, codecov sees no session for the producer-boundary slow path or the `refreshHeaderSupport` disable arms.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What does this PR do?
Adds a
kafka.messages.offsetstag to the kafkajs producer span, carrying the list of(partition, start_offset)pairs the broker returned fromproducer.send.start_offsetis the offset assigned to the first record sent to that partition (kafkajs'sbaseOffset).Example value for a batch that landed in three partitions:
```json
[{"partition":0,"start_offset":7},{"partition":1,"start_offset":2},{"partition":2,"start_offset":3}]
```
For single-record sends, the producer span additionally gets flat
kafka.partitionandkafka.message.offsettags — in that case the broker response describes exactly one record, so the flat tags are unambiguous and easy to filter on.No changes to consumer spans. The eachMessage consume span already carries
kafka.partitionandkafka.message.offset(per-record, exact); the eachBatch consume span is single-partition by construction, so its existingkafka.partitiontag is sufficient.Motivation
Inspired by dd-trace-java #11107, which adds per-record
partitionandoffsetto the producer span. The Java client emits one span persendcall (one record), so a single value is exact. kafkajs'sproducer.sendis batched — one span covers multiple records that may land in multiple partitions — so a list better reflects what actually happened.Combined with the already-tagged
kafka.cluster_id,kafka.topic, andkafka.batch_size, this lets a producer trace span be correlated to the exact set of partitions and starting offsets it produced to.Per-partition record counts are not in the broker response and aren't included; total batch size remains available via
kafka.batch_size.Additional Notes
🤖 Generated with Claude Code