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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
