Re: [PR] KAFKA-21197: Preserve original record headers in share group DLQ records. [kafka]

["AndrewJSchofield (via GitHub)" <gi...@apache.org>]

AndrewJSchofield commented on code in PR #23659:
URL: https://github.com/apache/kafka/pull/23659#discussion_r4206683036


##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQRecordHelper.java:
##########
@@ -61,6 +62,19 @@ private ShareGroupDLQRecordHelper() {}
     protected static final String HEADER_DLQ_ERRORS_DELIVERY_COUNT = "__dlq.errors.delivery.count";
     protected static final String HEADER_DLQ_ERRORS_MESSAGE = "__dlq.errors.message";
 
+    // DLQ context header keys added unconditionally by headers() below. An original record header
+    // reusing one of these keys is dropped (not merely shadowed) so the DLQ-computed value is the
+    // only one present - not just the one a last-header-wins lookup happens to find. The delivery
+    // count and cause/message headers are deliberately excluded here: they are only added when
+    // deliveryCount/cause are actually present (see headers() below), and dropping an original
+    // header under one of those keys when no replacement is being added would lose it outright.
+    private static final Set<String> RESERVED_DLQ_HEADER_KEYS = Set.of(
+            HEADER_DLQ_ERRORS_TOPIC,
+            HEADER_DLQ_ERRORS_PARTITION,
+            HEADER_DLQ_ERRORS_OFFSET,
+            HEADER_DLQ_ERRORS_GROUP

Review Comment:
   I think you should be dropping the other two in all cases too. I think the most likely scenario in which a user would create a record with these headers used is to forward a record from one DLQ topic back onto a normal topic, and then it gets DLQed again. So, preserving the ancient headers would be misleading.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org