GH-2155: DeadLetterPublishingRecoverer Improvement

Resolves #2155

Co-authored-by: Viacheslav Chernyshev <[email protected]>

- Add a `BitSet` property to suppress individual standard headers
- Support multiple `headersFunction`
- Allow complete customization of exception headers
- Add `setHeadersFunction` to the DLPR factory for retryable topics

* Fix typo in doc.

* Rework enum usage; other refactoring; add getters to `HeaderNames`.

* Fix javadocs.

* Doc polishing.

* Change BitSet to EnumSet.

* Fix bogus import.

* Defend against immutable headers; only create new when detected.

* Fix NOSONAR typo.

**Cherry-pick to `2.8.x`**

Author: garyrussell

spring-kafka-docs/src/main/asciidoc/kafka.adoc

@@ -5641,7 +5641,7 @@ Key exceptions are only caused by `DeserializationException` s so there is no `D
 There are two mechanisms to add more headers.
 
 1. Subclass the recoverer and override `createProducerRecord()` - call `super.createProducerRecord()` and add more headers.
-2. Provide a `BiFunction` to receive the consumer record and exception, returning a `Headers` object; headers from there will be copied to the final producer record.
+2. Provide a `BiFunction` to receive the consumer record and exception, returning a `Headers` object; headers from there will be copied to the final producer record; also see <<dlpr-headers>>.
 Use `setHeadersFunction()` to set the `BiFunction`.
 
 The second is simpler to implement but the first has more information available, including the already assembled standard headers.
@@ -5736,6 +5736,34 @@ The reason for the two properties is because, while you might want to retain onl
 
 `appendOriginalHeaders` is applied to all headers named `*ORIGINAL*` while `stripPreviousExceptionHeaders` is applied to all headers named `*EXCEPTION*`.
 
+Starting with version 2.8.4, you now can control which of the standard headers will be added to the output record.
+See the `enum HeadersToAdd` for the generic names of the (currently) 10 standard headers that are added by default (these are not the actual header names, just an abstraction; the actual header names are set up by the `getHeaderNames()` method which subclasses can override.
+
+To exclude headers, use the `excludeHeaders()` method; for example, to suppress adding the exception stack trace in a header, use:
+
+====
+[source, java]
+----
+DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(template);
+recoverer.excludeHeaders(HeaderNames.HeadersToAdd.EX_STACKTRACE);
+----
+====
+
+In addition, you can completely customize the addition of exception headers by adding an `ExceptionHeadersCreator`; this also disables all standard exception headers.
+
+====
+[source, java]
+----
+DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(template);
+recoverer.setExceptionHeadersCreator((kafkaHeaders, exception, isKey, headerNames) -> {
+    kafkaHeaders.add(new RecordHeader(..., ...));
+});
+----
+====
+
+Also starting with version 2.8.4, you can now provide multiple headers functions, via the `addHeadersFunction` method.
+This allows additional functions to apply, even if another function  has already been registered, for example, when using <<retry-topic>>.
+
 Also see <<retry-headers>> with <<retry-topic>>.
 
 [[exp-backoff]]

spring-kafka-docs/src/main/asciidoc/retrytopic.adoc

@@ -433,6 +433,8 @@ DeadLetterPublishingRecovererFactory factory(DestinationTopicResolver resolver)
 ----
 ====
 
+Starting with version 2.8.4, if you wish to add custom headers (in addition to the retry information headers added by the factory, you can add a `headersFunction` to the factory - `factory.setHeadersFunction((rec, ex) -> { ... })`
+
 [[retry-topic-combine-blocking]]
 ==== Combining blocking and non-blocking retries
 

spring-kafka-docs/src/main/asciidoc/whats-new.adoc

@@ -83,6 +83,8 @@ See <<delegating-serialization>> for more information.
 
 The property `stripPreviousExceptionHeaders` is now `true` by default.
 
+There are now several techniques to customize which headers are added to the output record.
+
 See <<dlpr-headers>> for more information.
 
 [[x28-retryable-topics-changes]]