refactor: simplify MSK header parsing per review

- Guard event.records with an early return instead of an empty fallback
- Wrap the record loop in a single try/catch so errors log once
- Return null from getParsedRecordHeaders when nothing decodes, lazily
  creating the headers map
- Drop per-byte validation and rely on Buffer.from

Author: lucassarcanjo

src/trace/context/extractors/msk.spec.ts

@@ -79,20 +79,18 @@ describe("MSKEventTraceExtractor", () => {
     expect(extract).toHaveBeenNthCalledWith(2, { "x-datadog-parent-id": "456" });
   });
 
-  it.each([null, "123", [256], [-1], [1.5], ["49"], [null], [true], { "0": 49 }].map((value) => [value]))(
-    "ignores invalid byte values: %j",
-    (value) => {
-      extract.mockReturnValue(spanContext);
-      expect(
-        extractor.extract(event({ "topic-0": [{ headers: [{ invalid: value }, ...record(datadogHeaders).headers] }] })),
-      ).toBe(spanContext);
-      expect(extract).toHaveBeenCalledWith(datadogHeaders);
-    },
-  );
+  it.each([null, "123", { "0": 49 }].map((value) => [value]))("skips non-array header values: %j", (value) => {
+    extract.mockReturnValue(spanContext);
+    expect(
+      extractor.extract(event({ "topic-0": [{ headers: [{ invalid: value }, ...record(datadogHeaders).headers] }] })),
+    ).toBe(spanContext);
+    expect(extract).toHaveBeenCalledWith(datadogHeaders);
+  });
 
   it.each([
     undefined,
     null,
+    "invalid",
     {},
     { "topic-0": [] },
     { "topic-0": null },
@@ -103,13 +101,12 @@ describe("MSKEventTraceExtractor", () => {
     expect(extract).not.toHaveBeenCalled();
   });
 
-  it("continues after an extraction error", () => {
-    extract
-      .mockImplementationOnce(() => {
-        throw new Error("invalid carrier");
-      })
-      .mockReturnValueOnce(spanContext);
-    expect(extractor.extract(event({ "topic-0": [record(w3cHeaders), record(datadogHeaders)] }))).toBe(spanContext);
+  it("returns null after an extraction error", () => {
+    extract.mockImplementationOnce(() => {
+      throw new Error("invalid carrier");
+    });
+    expect(extractor.extract(event({ "topic-0": [record(w3cHeaders), record(datadogHeaders)] }))).toBeNull();
+    expect(extract).toHaveBeenCalledTimes(1);
   });
 
   it("returns null when the tracer cannot extract context", () => {

src/trace/context/extractors/msk.ts

@@ -9,38 +9,44 @@ export class MSKEventTraceExtractor implements EventTraceExtractor {
   constructor(private tracerWrapper: TracerWrapper) {}
 
   extract(event: MSKEvent): SpanContextWrapper | null {
-    // A Lambda span can have only one parent. Use the first record with valid
-    // trace context, without combining headers from different records.
-    for (const records of Object.values(event.records ?? {})) {
-      if (!Array.isArray(records)) continue;
-      for (const record of records) {
-        try {
+    if (!event?.records) {
+      logDebug("Failed to extract trace context from MSK event");
+      return null;
+    }
+
+    try {
+      // A Lambda span can have only one parent. Use the first record with valid
+      // trace context, without combining headers from different records.
+      for (const records of Object.values(event.records)) {
+        if (!Array.isArray(records)) continue;
+        for (const record of records) {
           const headers = this.getParsedRecordHeaders(record);
-          if (Object.keys(headers).length === 0) continue;
+          if (!headers) continue;
           const traceContext = this.tracerWrapper.extract(headers);
           if (traceContext) {
             logDebug("Extracted trace context from MSK event");
             return traceContext;
           }
-        } catch (error) {
-          handleExtractionError(error, "MSK");
         }
       }
+    } catch (error) {
+      handleExtractionError(error, "MSK");
     }
+
     logDebug("Failed to extract trace context from MSK event");
     return null;
   }
 
-  private getParsedRecordHeaders(record: MSKRecord): Record<string, string> {
-    const headers: Record<string, string> = Object.create(null);
-    if (!Array.isArray(record?.headers)) return headers;
+  private getParsedRecordHeaders(record: MSKRecord): Record<string, string> | null {
+    if (!Array.isArray(record?.headers)) return null;
 
+    let headers: Record<string, string> | null = null;
     for (const entry of record.headers) {
       if (!entry || typeof entry !== "object" || Array.isArray(entry)) continue;
       for (const [name, value] of Object.entries(entry)) {
         // MSK serializes Kafka header bytes as integer arrays, not base64.
-        // Validate before decoding because Buffer.from silently coerces invalid bytes.
-        if (Array.isArray(value) && value.every((byte) => Number.isInteger(byte) && byte >= 0 && byte <= 255)) {
+        if (Array.isArray(value)) {
+          headers ??= {};
           headers[name.toLowerCase()] = Buffer.from(value).toString("utf8");
         }
       }