lianetm commented on code in PR #13898:
URL: https://github.com/apache/kafka/pull/13898#discussion_r1252190359
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetFetcherUtils.java:
##########
@@ -198,6 +208,11 @@ void validatePositionsOnMetadataChange() {
}
Map<TopicPartition, Long> getOffsetResetTimestamp() {
+ // Raise exception from previous offset fetch if there is one
+ RuntimeException exception =
cachedListOffsetsException.getAndSet(null);
Review Comment:
This should keep the existing logic for the `resetPositionsIfNeeded`, only
moving the exception check to this common functionality
`getOffsetResetTimestamp`(to be reused), and that is already called from the
reset (and only from there for now).
Existing logic checks exception and calls this `getOffsetResetTimestamp`
[here](https://github.com/lianetm/kafka/blob/d5dafe22fed244d25cca8839c41221fab87d367e/clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetFetcher.java#L119C1-L125C104)
New logic performs the same check but as part of the
`getOffsetResetTimestamp` called from the reset.
There is also a test `testRestOffsetsAuthorizationFailure` for this logic,
that passes with the new changes, ensuring
[here](https://github.com/apache/kafka/blob/701f924352da1225a881f0f78f19ddf51485030a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/OffsetFetcherTest.java#L579-L584)
that the behaviour remains as it was before.
Please let me know if I'm missing some detail here...
--
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
