GH-3121: Fix NPE when acknowledging a DLQ record without an Acknowledgment header

[Sanskar121543]

What

Fixes #3121.

With a transactional producer, ackMode=MANUAL_IMMEDIATE, and enableDlq=true, an error during processing can cause the binder to throw:

java.lang.NullPointerException: Cannot invoke "org.springframework.kafka.support.Acknowledgment.acknowledge()" because the return value of "org.springframework.messaging.MessageHeaders.get(Object, java.lang.Class)" is null
	at org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder$DlqSender.sendToDlq(...)

instead of acknowledging the record after it's sent to the DLQ. Because the ack never happens, the record is redelivered and sent to the DLQ again, producing many duplicate DLQ entries for a single failure.

Root cause

KafkaMessageChannelBinder.handleRecordForDlq derives the MessageHeaders passed into DlqSender#sendToDlq from ((ErrorMessage) message).getOriginalMessage().getHeaders(), falling back to message.getHeaders() when there is no original message. In the reported scenario (transactional producer + manual ack), the message handed to the DLQ error handler is an ErrorMessage with no original message, so the fallback message.getHeaders() is used - and that does not carry the KafkaHeaders.ACKNOWLEDGMENT header the original consumer record had.

DlqSender#sendToDlq's finally block doesn't guard against this and unconditionally does:

messageHeaders.get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class).acknowledge();

which NPEs as soon as that header is absent, for any MANUAL/MANUAL_IMMEDIATE ack mode record that reaches the DLQ this way.

Fix

Null-check the Acknowledgment before calling acknowledge(). When it's missing, log a warning instead of throwing, so the DLQ send itself still completes cleanly rather than blowing up with an unhandled NPE. Behavior is unchanged when the Acknowledgment header is present (the common case) or when the ack mode isn't MANUAL/MANUAL_IMMEDIATE.

Testing

Added three unit tests to KafkaBinderUnitTests that invoke KafkaMessageChannelBinder$DlqSender#sendToDlq directly (via reflection, following the existing pattern in that test class) with a mocked KafkaTemplate, so no broker is needed:

• sendToDlqDoesNotThrowWhenAcknowledgmentHeaderIsMissing - reproduces the reported NPE. I verified this test fails with the exact reported NullPointerException (same stack frame) against the unpatched code, and passes with the fix.
• sendToDlqStillAcknowledgesWhenAcknowledgmentHeaderIsPresent - confirms the normal case (Acknowledgment present) still calls acknowledge().
• sendToDlqDoesNotAcknowledgeWhenAckModeIsNotManual - confirms non-manual ack modes are unaffected.

Ran KafkaBinderUnitTests (10/10 passing) and a full mvn test-compile of the spring-cloud-stream-binder-kafka module to confirm no regressions elsewhere. I did not run the broker-backed integration suite (KafkaBinderTests, embedded-Kafka tests) locally due to runtime constraints, but the change is a narrowly-scoped null guard in a single finally block that cannot change behavior on any path where the Acknowledgment header is present.