feat(triggers): handle MSK events

[joeyzhao2018]

https://datadoghq.atlassian.net/browse/SLES-2739

In Kafka's wire protocol (KIP-82), header values are always byte[]. Every Kafka client library enforces this:

Tracer	Injection code	Mechanism
dd-trace-java	headers.add(key, value.getBytes(UTF_8))	String.getBytes() → byte[]
dd-trace-go	Value: []byte(val)	Go type conversion → []byte
dd-trace-dotnet	_headers.Add(name, Encoding.UTF8.GetBytes(value))	UTF8.GetBytes() → byte[]

All three tracers accept string trace context values from the propagation layer, convert to UTF-8 bytes at the carrier adapter boundary, and hand byte[] to the Kafka client.
This isn't a quirk of Java's getBytes() — it's the only way Kafka headers work.

What MSK Lambda does

When MSK triggers a Lambda, AWS serializes the Kafka record to JSON. Since header values are byte[] on the wire, AWS encodes them as decimal byte values. However, the exact JSON
shape depends on the Lambda runtime:

• Array format (observed in the existing msk_event.json testing payloads, i didn't change the support for this to be safe): byte values as a JSON array of integers
  "headers": [{"x-datadog-trace-id": [51, 54, 57, ...]}]
• Object format (observed with the Java Lambda runtime): both the records list and the per-header byte values are JSON objects with numeric string keys, and byte values are
  decimal strings
  "records": {
  "topic-0": {
  "0": {
  "headers": {
  "0": {"someOtherHeader": ["70", "114", ...]},
  "2": {"x-datadog-trace-id": {"0":"52","1":"54",...}},
  "4": {"x-datadog-sampling-priority": ["49"]}
  }
  }
  }
  }
• Note that Datadog headers can appear at any index — non-instrumentation headers may precede them.

What's the difference between the msk_event.json and the newly added msk_event_with_headers.json here?

• msk_event.json represents a standard MSK trigger where the producer didn't attach any Kafka headers — i.e. no Datadog tracer was running on the producer side (or it's a non-instrumented producer like a raw Kafka client, a Kinesis Firehose delivery stream, or a schema-registry message). In those cases Lambda still delivers the event but with "headers": []. It's also the format you get when testing MSK triggers manually in the AWS console, which doesn't inject headers. ( source: Claude Code)
• msk_event_with_headers.json reflects the real-world object format produced by the Java Lambda runtime, with a producer instrumented with a Datadog tracer injecting trace context
  as Kafka headers. It includes non-Datadog headers at lower indices to verify that the carrier extraction correctly finds Datadog headers regardless of their position. (source: I did a real world example and below is the evidence of testing)

[joeyzhao2018]

why the changes here?

it's just trying to fix the test failures. I also put this change in a separate pr here #1099 if you prefer that way.

[Copilot]

Pull request overview

This PR adds support for MSK (Managed Streaming for Apache Kafka) events where records and headers are delivered in a JSON object format with numeric string keys (observed with the Java Lambda runtime), in addition to the existing array format. It extracts Datadog trace context from Kafka headers encoded as byte arrays (decimal byte values), enabling distributed tracing for MSK-triggered Lambda functions.

Changes:

• Added bytes_from_header_value and carrier_from_headers functions to decode Kafka header byte values (both array and object formats) and extract trace propagation headers into a carrier map.
• Updated MSKEvent::new and is_match to handle both array and object record formats from different Lambda runtimes.
• Added a new test payload (msk_event_with_headers.json) and tests validating deserialization and carrier extraction for the object-format MSK events.

Reviewed changes

Copilot reviewed 2 out of 3 changed files in this pull request and generated 2 comments.

File	Description
bottlecap/src/lifecycle/invocation/triggers/msk_event.rs	Core logic: header decoding, carrier extraction, updated new/is_match for object-format records, new tests
bottlecap/tests/payloads/msk_event_with_headers.json	New test payload representing Java runtime MSK event with Datadog trace headers
bottlecap/Cargo.lock	Routine dependency updates (itertools, quinn-proto)

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

You can also share your feedback on Copilot code review. Take the survey.