fix(kafkajs): skip header injection on brokers below Produce v3#8390
Merged
BridgeAR merged 2 commits intoMay 19, 2026
Conversation
Contributor
Overall package sizeSelf size: 5.88 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: 5a1d7ce | Docs | Datadog PR Page | Give us feedback! |
BenchmarksBenchmark execution time: 2026-05-18 19:20:59 Comparing candidate commit 5a1d7ce in PR branch Found 0 performance improvements and 0 performance regressions! Performance is the same for 1491 metrics, 102 unstable metrics. |
BridgeAR
force-pushed
the
BridgeAR/2026-05-09-kafka-read-clusterid-from-metadata
branch
from
May 11, 2026 22:22
a1ae644 to
0d852ca
Compare
Base automatically changed from
BridgeAR/2026-05-09-kafka-read-clusterid-from-metadata
to
master
May 12, 2026 13:49
The Produce request didn't carry headers until v3 (Kafka 0.11), so injecting on older brokers either wastes work or trips `UNKNOWN_SERVER_ERROR` on a per-leader version mismatch in mixed-version clusters. Read `cluster.brokerPool.versions[0].maxVersion` once on first send and disable injection proactively when the broker negotiated <v3; the existing `KafkaJSProtocolError UNKNOWN` handler stays as the safety net for mixed-version clusters where the seed broker advertised v3+ but the leader couldn't parse the headers. `refreshHeaderSupport` is a `let` that swaps to module-level `noop` at both disable sites, so subsequent sends skip the helper call entirely. The reactive `disableHeaderInjection` moves from the module-level `disabledHeaderWeakSet` to a per-producer closure so the lifetime matches the producer it describes. Refs: #8253
BridgeAR
force-pushed
the
BridgeAR/2026-05-09-kafka-skip-injection-pre-v3
branch
from
May 12, 2026 14:37
681b6c4 to
9792e32
Compare
BridgeAR
marked this pull request as ready for review
May 12, 2026 14:38
tlhunter
reviewed
May 12, 2026
| // be <0.11; lying locally is enough to drive the proactive | ||
| // header-support check. | ||
| const cluster = clientToCluster.get(producer) | ||
| cluster.brokerPool.versions[0].maxVersion = 2 |
Member
There was a problem hiding this comment.
Shouldn't we actually spin up a Docker container with an old version of Kafka to verify it works as expected instead of just mocking this value?
Member
Author
There was a problem hiding this comment.
I think the mock is better, since such old Kafka brokers are from 2017 :D
So having the mock here seems the best approach, no?
tlhunter
approved these changes
May 18, 2026
Merged
dd-octo-sts Bot
pushed a commit
that referenced
this pull request
May 19, 2026
The Produce request didn't carry headers until v3 (Kafka 0.11), so injecting on older brokers either wastes work or trips `UNKNOWN_SERVER_ERROR` on a per-leader version mismatch in mixed-version clusters. Read `cluster.brokerPool.versions[0].maxVersion` once on first send and disable injection proactively when the broker negotiated <v3; the existing `KafkaJSProtocolError UNKNOWN` handler stays as the safety net for mixed-version clusters where the seed broker advertised v3+ but the leader couldn't parse the headers. `refreshHeaderSupport` is a `let` that swaps to module-level `noop` at both disable sites, so subsequent sends skip the helper call entirely. The reactive `disableHeaderInjection` moves from the module-level `disabledHeaderWeakSet` to a per-producer closure so the lifetime matches the producer it describes.
BridgeAR
added a commit
that referenced
this pull request
May 19, 2026
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
BridgeAR
added a commit
that referenced
this pull request
May 19, 2026
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
pabloerhard
pushed a commit
that referenced
this pull request
May 20, 2026
The Produce request didn't carry headers until v3 (Kafka 0.11), so injecting on older brokers either wastes work or trips `UNKNOWN_SERVER_ERROR` on a per-leader version mismatch in mixed-version clusters. Read `cluster.brokerPool.versions[0].maxVersion` once on first send and disable injection proactively when the broker negotiated <v3; the existing `KafkaJSProtocolError UNKNOWN` handler stays as the safety net for mixed-version clusters where the seed broker advertised v3+ but the leader couldn't parse the headers. `refreshHeaderSupport` is a `let` that swaps to module-level `noop` at both disable sites, so subsequent sends skip the helper call entirely. The reactive `disableHeaderInjection` moves from the module-level `disabledHeaderWeakSet` to a per-producer closure so the lifetime matches the producer it describes.
BridgeAR
added a commit
that referenced
this pull request
May 22, 2026
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
2 tasks
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
2 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.
Summary
The Produce request didn't carry headers until v3 (Kafka 0.11), so injecting on older brokers either wastes work or trips
UNKNOWN_SERVER_ERRORon a per-leader version mismatch in mixed-version clusters.cluster.brokerPool.versions[0].maxVersiononce on first send and disable injection proactively when the broker negotiated <v3.KafkaJSProtocolError UNKNOWNhandler stays as the safety net for mixed-version clusters where the seed broker advertised v3+ but the leader couldn't parse the headers.refreshHeaderSupportis aletthat swaps to module-levelnoopat both disable sites, so subsequent sends skip the helper call entirely. The reactivedisableHeaderInjectionmoves from the module-leveldisabledHeaderWeakSetto a per-producer closure so the lifetime matches the producer it describes.Test plan
./node_modules/.bin/mocha packages/datadog-instrumentations/test/helpers/kafka.spec.js— unit tests forbrokerSupportsMessageHeaders(boundary at maxVersion 2 / 3 / 9 / undefined).PLUGINS=kafkajs SPEC=index npm run test:plugins— pins that downgrading the broker pool's negotiated Produce version to 2 trips the proactive disable path; the cloned messages reach the channel withoutheaders: {}seeding.Stacked on top of #8389.
Refs: #8253