…st disable

Two coordinated changes addressing review feedback on the producer wrap:

1. Replace the Symbol-keyed cluster reference (`producer[kCluster] = ...`)
   with a `WeakMap` exported from `helpers/kafka`. Symbols still mutate the
   kafkajs producer/consumer: they leak through `Reflect.ownKeys` /
   `Object.getOwnPropertySymbols`, force a hidden-class transition on the
   kafkajs-returned object, and trip on frozen producers. The side-table
   keeps the kafkajs object untouched. The integration test that downgrades
   the broker version reads the cluster through the same map instead of
   `Object.getOwnPropertySymbols`.

2. `refreshHeaderSupport` becomes a `let` that swaps to a module-level
   `noop` at both disable sites — proactive Produce <v3 detection and the
   reactive `KafkaJSProtocolError UNKNOWN` handler — so subsequent sends
   skip the helper call entirely.

Drive-by fix:

* Blank line between adjacent `it` blocks in `index.spec.js`.

Refs: #8253

Match the wrapper shape already used in the kafkajs index spec for the
same assertion.

Refs: #8253

Brokers that fail when any header field is present — kafkajs's
`KafkaJSProtocolError UNKNOWN`, `@confluentinc/kafka-javascript`'s
`KafkaJSError ERR_UNKNOWN`, kafkajs negotiating Produce <v3 — also reject
empty `headers: {}`. The disable path was meant to recover for those brokers,
but `cloneMessagesForInjection` still seeded `{}` on every cloned message so
the client kept seeing a header field and the broker kept rejecting it.

Refs: #8253

…nsureHeaders)

The two clone helpers in `helpers/kafka.js` differed only in what they did
with messages that arrived without a `headers` field: one preserved the
absence, the other seeded `headers: {}` for the producer plugin to inject
into. Pass that as a second argument to a single `cloneMessages` and let
the call site decide based on `disableHeaderInjection`. Same hot-path cost,
one fewer code path to keep in sync.

Refs: #8253 (comment)

…st disable

Two coordinated changes addressing review feedback on the producer wrap:

1. Replace the Symbol-keyed cluster reference (`producer[kCluster] = ...`)
   with a `WeakMap` exported from `helpers/kafka`. Symbols still mutate the
   kafkajs producer/consumer: they leak through `Reflect.ownKeys` /
   `Object.getOwnPropertySymbols`, force a hidden-class transition on the
   kafkajs-returned object, and trip on frozen producers. The side-table
   keeps the kafkajs object untouched. The integration test that downgrades
   the broker version reads the cluster through the same map instead of
   `Object.getOwnPropertySymbols`.

2. `refreshHeaderSupport` becomes a `let` that swaps to a module-level
   `noop` at both disable sites — proactive Produce <v3 detection and the
   reactive `KafkaJSProtocolError UNKNOWN` handler — so subsequent sends
   skip the helper call entirely.

Drive-by fix:

* Blank line between adjacent `it` blocks in `index.spec.js`.

Refs: #8253

Match the wrapper shape already used in the kafkajs index spec for the
same assertion.

Refs: #8253

Brokers that fail when any header field is present — kafkajs's
`KafkaJSProtocolError UNKNOWN`, `@confluentinc/kafka-javascript`'s
`KafkaJSError ERR_UNKNOWN`, kafkajs negotiating Produce <v3 — also reject
empty `headers: {}`. The disable path was meant to recover for those brokers,
but `cloneMessagesForInjection` still seeded `{}` on every cloned message so
the client kept seeing a header field and the broker kept rejecting it.

Refs: #8253

…nsureHeaders)

The two clone helpers in `helpers/kafka.js` differed only in what they did
with messages that arrived without a `headers` field: one preserved the
absence, the other seeded `headers: {}` for the producer plugin to inject
into. Pass that as a second argument to a single `cloneMessages` and let
the call site decide based on `disableHeaderInjection`. Same hot-path cost,
one fewer code path to keep in sync.

Refs: #8253 (comment)

Both kafkajs's producer wrap and the `@confluentinc/kafka-javascript`
KafkaJS-compat producer wrap mutated the caller's `message.headers` in
place. The kafkajs boundary seeded `headers: {}` on every message before
the plugin injected trace context; the confluent KafkaJS-compat path
handed `payload.messages` straight to the underlying client, which then
wrote `headers: null` on messages without headers. Either path leaves
the caller holding the side effects of someone else's instrumentation.

The boundary now hands kafkajs and the underlying confluent client a
shallow clone of the messages array via a single `cloneMessages`
helper. With injection enabled the helper seeds `headers: {}` on the
clone for the plugin to inject into; with injection disabled (after the
broker reported `KafkaJSProtocolError UNKNOWN` /
`KafkaJSError ERR_UNKNOWN`) the helper preserves the absence of
`headers` so brokers that reject any header field can recover.

Refs: #5270
Refs: #8253

The cluster id discovery used a separate `kafka.admin()` connection whose
`describeCluster()` rejection wasn't awaited, surfacing as
`process.unhandledRejection` whenever the admin handshake failed (auth,
broker restart, network blip). Read
`cluster.brokerPool.metadata.clusterId` instead, priming it on first send
via `cluster.refreshMetadataIfNecessary()`; kafkajs's `sharedPromiseTo`
collapses our call with its internal one, so total latency is unchanged.

The cluster reference for each producer/consumer is captured at creation
time via two new `addHook` entries on `src/producer/index.js` /
`src/consumer/index.js` and stored in a `WeakMap` exported from
`helpers/kafka`. The kafkajs object stays untouched — no Symbol-keyed
property to leak through `Reflect.ownKeys`, no string-keyed underscore
for user serializers to pick up, and the entry drops as soon as the
producer is GC'd.

Refs: #5270
Refs: #8253

The cluster id discovery used a separate `kafka.admin()` connection whose
`describeCluster()` rejection wasn't awaited, surfacing as
`process.unhandledRejection` whenever the admin handshake failed (auth,
broker restart, network blip). Read
`cluster.brokerPool.metadata.clusterId` instead, priming it on first send
via `cluster.refreshMetadataIfNecessary()`; kafkajs's `sharedPromiseTo`
collapses our call with its internal one, so total latency is unchanged.

The cluster reference for each producer/consumer is captured at creation
time via two new `addHook` entries on `src/producer/index.js` /
`src/consumer/index.js` and stored in a `WeakMap` exported from
`helpers/kafka`. The kafkajs object stays untouched — no Symbol-keyed
property to leak through `Reflect.ownKeys`, no string-keyed underscore
for user serializers to pick up, and the entry drops as soon as the
producer is GC'd.

Refs: #5270
Refs: #8253

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

The cluster id discovery used a separate `kafka.admin()` connection whose
`describeCluster()` rejection wasn't awaited, surfacing as
`process.unhandledRejection` whenever the admin handshake failed (auth,
broker restart, network blip). Read
`cluster.brokerPool.metadata.clusterId` instead, priming it on first send
via `cluster.refreshMetadataIfNecessary()`; kafkajs's `sharedPromiseTo`
collapses our call with its internal one, so total latency is unchanged.

The cluster reference for each producer/consumer is captured at creation
time via two new `addHook` entries on `src/producer/index.js` /
`src/consumer/index.js` and stored in a `WeakMap` exported from
`helpers/kafka`. The kafkajs object stays untouched — no Symbol-keyed
property to leak through `Reflect.ownKeys`, no string-keyed underscore
for user serializers to pick up, and the entry drops as soon as the
producer is GC'd.

Refs: #5270
Refs: #8253

Both kafkajs's producer wrap and the `@confluentinc/kafka-javascript`
KafkaJS-compat producer wrap mutated the caller's `message.headers` in
place. The kafkajs boundary seeded `headers: {}` on every message before
the plugin injected trace context; the confluent KafkaJS-compat path
handed `payload.messages` straight to the underlying client, which then
wrote `headers: null` on messages without headers. Either path leaves
the caller holding the side effects of someone else's instrumentation.

The boundary now hands kafkajs and the underlying confluent client a
shallow clone of the messages array via a single `cloneMessages`
helper. With injection enabled the helper seeds `headers: {}` on the
clone for the plugin to inject into; with injection disabled (after the
broker reported `KafkaJSProtocolError UNKNOWN` /
`KafkaJSError ERR_UNKNOWN`) the helper preserves the absence of
`headers` so brokers that reject any header field can recover.

Refs: #5270
Refs: #8253

The cluster id discovery used a separate `kafka.admin()` connection whose
`describeCluster()` rejection wasn't awaited, surfacing as
`process.unhandledRejection` whenever the admin handshake failed (auth,
broker restart, network blip). Read
`cluster.brokerPool.metadata.clusterId` instead, priming it on first send
via `cluster.refreshMetadataIfNecessary()`; kafkajs's `sharedPromiseTo`
collapses our call with its internal one, so total latency is unchanged.

The cluster reference for each producer/consumer is captured at creation
time via two new `addHook` entries on `src/producer/index.js` /
`src/consumer/index.js` and stored in a `WeakMap` exported from
`helpers/kafka`. The kafkajs object stays untouched — no Symbol-keyed
property to leak through `Reflect.ownKeys`, no string-keyed underscore
for user serializers to pick up, and the entry drops as soon as the
producer is GC'd.

Refs: #5270
Refs: #8253