KAFKA-20462: Preserve original source-record headers in ErrorHandlerContext

[1fanwang]

Problem

ErrorHandlerContext#headers() is documented to expose "the headers of the current source record". Both error paths in Streams violate that contract today: a user-supplied Deserializer is allowed to mutate the live Headers reference (e.g. a LargeMessageDeserializer that removes its serde-internal marker header so it doesn't leak downstream), and the same mutable reference flows into the DefaultErrorHandlerContext that the DeserializationExceptionHandler and ProcessingExceptionHandler see. ExceptionHandlerUtils#buildDeadLetterQueueRecord then copies the post-mutation headers into the DLQ record, so a DLQ record produced from a mutating-Deserializer pipeline can no longer be deserialized by the same Deserializer on a replay (its marker header is gone).

This complements #21370 (which made buildDeadLetterQueueRecord copy context.headers() into the DLQ record): the DLQ now copies something, but per KAFKA-20462 that something must be the original source-record headers, which is what this PR provides.

Evidence

• Reproducer (deserialization-error path): a Deserializer that calls headers.remove("source-only") and then throws causes ErrorHandlerContext#headers() in the DeserializationExceptionHandler to return headers without "source-only". New test RecordDeserializerTest#shouldExposeOriginalSourceRecordHeadersToErrorHandlerWhenDeserializerMutatesHeaders is verified to fail before this fix and pass after.
• Reproducer (processing-error path): a ProcessorRecordContext where the live headers were mutated to empty but sourceRawHeaders carries the original "source-only" header. With the live headers fed through to the error context, ErrorHandlerContext#headers() returns the empty (mutated) view; with sourceRawHeaders fed through, it returns the original. New test ProcessorNodeTest#shouldExposeSourceRawHeadersToProcessingExceptionHandlerWhenDeserializerMutatedHeaders covers this.

Fix

Single snapshot per record, taken in RecordQueue#updateHead (and the analogous spots in GlobalStateUpdateTask / GlobalStateManagerImpl) before the user-supplied Deserializer runs. The snapshot is propagated along two paths:

1. As an explicit parameter to RecordDeserializer#deserialize so that, when deserialization throws, RecordDeserializer#handleDeserializationFailure builds DefaultErrorHandlerContext with the snapshot rather than the (possibly mutated) live reference. (This is the path my earlier commit on this branch addressed.)
2. On a new transient sourceRawHeaders field on StampedRecord and ProcessorRecordContext, mirroring the existing sourceRawKey / sourceRawValue lifecycle: not serialized, cleared by freeRawRecord(). ProcessorNode reads the snapshot from the ProcessorRecordContext (cast through a small helper, so the public RecordContext interface is unchanged — no KIP) when constructing the DefaultErrorHandlerContext for the ProcessingExceptionHandler.

The downstream Record passed to processors still carries the live (possibly mutated) Headers reference, so the LargeMessageDeserializer pattern — Deserializers that legitimately mutate headers so processors don't see serde-internal data — continues to work unchanged. Only ErrorHandlerContext#headers() is changed to honor its documented contract.

Cost: one RecordHeaders copy (small ArrayList<Header> clone, typically 0–3 entries) per record on the deserialization path. Negligible relative to the existing per-record allocations in RecordQueue / RecordDeserializer.

Compatibility

• Public API: unchanged. No additions to RecordContext or ErrorHandlerContext. sourceRawHeaders lives only on the internal ProcessorRecordContext and is read via a type-narrowing cast inside ProcessorNode. No KIP needed.
• Wire format: unchanged. sourceRawHeaders is transient (like sourceRawKey / sourceRawValue), not serialized into ProcessorRecordContext#serialize() and not surfaced in changelog topics or state-store buffers.
• User-visible behavior: ErrorHandlerContext#headers() now returns the original source-record headers (as documented) rather than the post-mutation reference. Users whose handlers depended on the buggy post-mutation behavior will see the unmutated headers instead.

Tests

• New RecordDeserializerTest#shouldExposeOriginalSourceRecordHeadersToErrorHandlerWhenDeserializerMutatesHeaders — DeserializationExceptionHandler path.
• New ProcessorNodeTest#shouldExposeSourceRawHeadersToProcessingExceptionHandlerWhenDeserializerMutatedHeaders — ProcessingExceptionHandler path.
• Both verified to fail on trunk and pass after this PR.
• Existing RecordDeserializerTest, ProcessorNodeTest, StreamTaskTest, RecordQueueTest, GlobalStateTaskTest, GlobalStateManagerImplTest, ProcessingExceptionHandlerIntegrationTest, ExceptionHandlerUtilsTest all pass.
• :streams:checkstyleMain :streams:checkstyleTest :streams:spotbugsMain all green.

JIRA: https://issues.apache.org/jira/browse/KAFKA-20462