GH-2155: DeadLetterPublishingRecoverer Improvement #2161

garyrussell · 2022-03-08T21:05:05Z

Resolves #2155

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

cherry-pick to 2.8.x

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

tomazfernandes · 2022-03-08T21:20:27Z

...ng-kafka/src/main/java/org/springframework/kafka/listener/DeadLetterPublishingRecoverer.java

-		maybeAddHeader(kafkaHeaders, this.headerNames.original.timestampTypeHeader,
-				record.timestampType().toString().getBytes(StandardCharsets.UTF_8));
-		if (ex instanceof ListenerExecutionFailedException) {
+		if (this.whichHeaders.get(HeaderNames.HeadersToAdd.TOPIC.getBit())) {


Wouldn't it maybe be clearer if this if statement was inside the maybeAddHeaders method, and we passed the enum as an argument?

tomazfernandes · 2022-03-08T21:26:41Z

...src/main/java/org/springframework/kafka/retrytopic/DeadLetterPublishingRecovererFactory.java

@@ -137,6 +150,9 @@ public DeadLetterPublishingRecoverer create() {
 		};

 		recoverer.setHeadersFunction((consumerRecord, e) -> addHeaders(consumerRecord, e, getAttempts(consumerRecord)));
+		if (this.headersFunction != null) {


Just to clarify, what I had in mind is maybe we could apply this headersFunction right here in the addHeaders method - this way we wouldn't need to make DLPR's header function composable and might make things simpler overall.

But also like this approach, which has the added benefit of warning the user if they set a headersFunction that would override RT's.

Yes; I was going to do it here too, but I had already added to the code the DLPR and it would have been duplicated here - I also thought this could be useful in other scenarios, where a library has a "standard" corporate DLPR bean in a jar and a user wants to use that, plus add more headers.

Sure, makes total sense. Thanks.

artembilan · 2022-03-08T21:40:03Z

...ng-kafka/src/main/java/org/springframework/kafka/listener/DeadLetterPublishingRecoverer.java

+			/**
+			 * The offset of the failed record.
+			 */
+			OFFSET(0),


Looks like just an ordinal() is enough.
I mean all those explicit values for this enum is fully aligned with an ordinal() of the enum:

/** * Returns the ordinal of this enumeration constant (its position * in its enum declaration, where the initial constant is assigned * an ordinal of zero). * * Most programmers will have no use for this method. It is * designed for use by sophisticated enum-based data structures, such * as {@link java.util.EnumSet} and {@link java.util.EnumMap}. * * @return the ordinal of this enumeration constant */ public final int ordinal() {

artembilan · 2022-03-08T21:40:23Z

...src/main/java/org/springframework/kafka/retrytopic/DeadLetterPublishingRecovererFactory.java

+	 * Set a function that creates additional headers for the output record, in addition to the standard
+	 * retry headers added by this factory.
+	 * @param headersFunction the function.
+	 * @since


...ng-kafka/src/main/java/org/springframework/kafka/listener/DeadLetterPublishingRecoverer.java

artembilan

I'm OK with this.

@tomazfernandes , @v-chernyshev ?

tomazfernandes · 2022-03-09T15:05:25Z

I'm OK with this.

@tomazfernandes , @v-chernyshev ?

Looks great to me. I didn’t know this BitSet class, works perfectly for this use case. Thanks.

v-chernyshev · 2022-03-09T15:11:10Z

...ng-kafka/src/main/java/org/springframework/kafka/listener/DeadLetterPublishingRecoverer.java

@@ -83,9 +88,11 @@

 	private final Function<ProducerRecord<?, ?>, KafkaOperations<?, ?>> templateResolver;

+	private final BitSet whichHeaders = new BitSet(10);


Can HeaderNames.HeadersToAdd.values().length be used here instead?

Hm. They say that a new array is allocated for every values() call, so I don't know whether it would make sense to also cache the length value...

One more question. What's the motivation for not using EnumSet<HeaderNames.HeadersToAdd> instead of BitSet? The documentation says that:

Enum sets are represented internally as bit vectors.

Then the whichHeaders set could be initialised with the help of allOf/noneOf static methods of EnumSet.

After reading EnumSet JavaDocs, I concur that this might be a better way for our use-case.

Agreed; I wasn't aware of it.

artembilan · 2022-03-09T15:27:08Z

...ng-kafka/src/main/java/org/springframework/kafka/listener/DeadLetterPublishingRecoverer.java

@@ -83,9 +88,11 @@

 	private final Function<ProducerRecord<?, ?>, KafkaOperations<?, ?>> templateResolver;

+	private final BitSet whichHeaders = new BitSet(10);


After reading EnumSet JavaDocs, I concur that this might be a better way for our use-case.

v-chernyshev · 2022-03-10T08:50:01Z

...ng-kafka/src/main/java/org/springframework/kafka/listener/DeadLetterPublishingRecoverer.java

@@ -45,6 +46,7 @@
 import org.springframework.kafka.KafkaException;
 import org.springframework.kafka.core.KafkaOperations;
 import org.springframework.kafka.core.ProducerFactory;
+import org.springframework.kafka.listener.DeadLetterPublishingRecoverer.HeaderNames.HeadersToAdd;


Just to confirm. There is no policy to avoid imports from the class that is declared in the same file, is there? There are two places that use HeadersToAdd rather than HeaderNames.HeadersToAdd, which causes this import statement to appear.

Resolves spring-projects#2155 - 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

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

v-chernyshev · 2022-03-10T14:35:24Z

...ng-kafka/src/main/java/org/springframework/kafka/listener/DeadLetterPublishingRecoverer.java

+				if (headers2 != null) {
+					headers2.forEach(headers1::add);
+				}


There's one minor point of concern with this implementation. Imagine that the original toCompose function does something similar to this:

Headers addHeaders(ConsumerRecord<...> record, Exception exception) { RecordHeaders extraHeaders = new RecordHeaders(); extraHeaders.add("key", ...); extraHeaders.setReadOnly(); // This is the important line. return extraHeaders; }

The returned value would be stored in headers1. headers2.forEach(headers1::add); would then fail with an IllegalStateException because headers1 can no longer be modified. A safer, albeit slightly more expensive, alternative could be:

// I believe the constructor is null-safe, but please double-check. RecordHeaders headers1 = new RecordHeaders(toCompose.apply(rec, ex)); Headers headers2 = headersFunction.apply(rec, ex); if (headers2 != null) { headers2.forEach(headers1::add); } return headers1;

Unusual; but thanks.

Yeah, I highly doubt anyone would actually do it, so I'm happy with the current implementation too.

artembilan

I'm OK with this.
Any other concerns before merging?

tomazfernandes · 2022-03-10T17:06:21Z

I'm OK with this.
Any other concerns before merging?

Not from me, thanks. And thanks @v-chernyshev for the thorough review and bringing in the EnumSet.

v-chernyshev · 2022-03-10T17:23:40Z

Not from me, thanks. And thanks @v-chernyshev for the thorough review and bringing in the EnumSet.

You are welcome, thanks for implementing this feature suggestion so quickly :) I do not have any further comments for the PR.

artembilan · 2022-03-10T17:40:38Z

... and cherry-picked to 2.8.x as f3f3eac after fixing conflicts.

v-chernyshev reviewed Mar 8, 2022

View reviewed changes

spring-kafka-docs/src/main/asciidoc/kafka.adoc Outdated Show resolved Hide resolved

tomazfernandes reviewed Mar 8, 2022

View reviewed changes

artembilan requested changes Mar 8, 2022

View reviewed changes

v-chernyshev reviewed Mar 8, 2022

View reviewed changes

...ng-kafka/src/main/java/org/springframework/kafka/listener/DeadLetterPublishingRecoverer.java Show resolved Hide resolved

artembilan approved these changes Mar 9, 2022

View reviewed changes

v-chernyshev reviewed Mar 9, 2022

View reviewed changes

artembilan requested changes Mar 9, 2022

View reviewed changes

artembilan approved these changes Mar 9, 2022

View reviewed changes

v-chernyshev reviewed Mar 10, 2022

View reviewed changes

garyrussell and others added 7 commits March 10, 2022 08:49

Fix typo in doc.

e5f9c1c

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

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

1f4ac71

Fix javadocs.

42a84da

Doc polishing.

3f7fb6a

Change BitSet to EnumSet.

9639262

Fix bogus import.

11e1bd9

garyrussell force-pushed the GH-2155 branch from e9512d9 to 11e1bd9 Compare March 10, 2022 13:50

v-chernyshev reviewed Mar 10, 2022

View reviewed changes

Defend against immutable headers; only create new when detected.

b3233d9

artembilan approved these changes Mar 10, 2022

View reviewed changes

Fix NOSONAR typo.

53619b9

artembilan approved these changes Mar 10, 2022

View reviewed changes

artembilan merged commit f83fc4b into spring-projects:main Mar 10, 2022

garyrussell deleted the GH-2155 branch March 11, 2022 15:59

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

GH-2155: DeadLetterPublishingRecoverer Improvement #2161

GH-2155: DeadLetterPublishingRecoverer Improvement #2161

garyrussell commented Mar 8, 2022 •

edited

Loading

tomazfernandes Mar 8, 2022

tomazfernandes Mar 8, 2022

garyrussell Mar 8, 2022

tomazfernandes Mar 8, 2022

artembilan Mar 8, 2022

artembilan Mar 8, 2022

artembilan left a comment

tomazfernandes commented Mar 9, 2022

v-chernyshev Mar 9, 2022

v-chernyshev Mar 9, 2022 •

edited

Loading

artembilan Mar 9, 2022

garyrussell Mar 9, 2022

artembilan Mar 9, 2022

v-chernyshev Mar 10, 2022

v-chernyshev Mar 10, 2022

garyrussell Mar 10, 2022

v-chernyshev Mar 10, 2022

artembilan left a comment

tomazfernandes commented Mar 10, 2022

v-chernyshev commented Mar 10, 2022 •

edited

Loading

artembilan commented Mar 10, 2022

		@@ -83,9 +88,11 @@

		private final Function<ProducerRecord<?, ?>, KafkaOperations<?, ?>> templateResolver;

		private final BitSet whichHeaders = new BitSet(10);

GH-2155: DeadLetterPublishingRecoverer Improvement #2161

GH-2155: DeadLetterPublishingRecoverer Improvement #2161

Conversation

garyrussell commented Mar 8, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

artembilan left a comment

Choose a reason for hiding this comment

tomazfernandes commented Mar 9, 2022

Choose a reason for hiding this comment

v-chernyshev Mar 9, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

artembilan left a comment

Choose a reason for hiding this comment

tomazfernandes commented Mar 10, 2022

v-chernyshev commented Mar 10, 2022 • edited Loading

artembilan commented Mar 10, 2022

garyrussell commented Mar 8, 2022 •

edited

Loading

v-chernyshev Mar 9, 2022 •

edited

Loading

v-chernyshev commented Mar 10, 2022 •

edited

Loading