mumrah commented on a change in pull request #8841:
URL: https://github.com/apache/kafka/pull/8841#discussion_r441742901
##########
File path:
clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
##########
@@ -1281,11 +1291,12 @@ private CompletedFetch
initializeCompletedFetch(CompletedFetch nextCompletedFetc
Optional<Integer> clearedReplicaId =
subscriptions.clearPreferredReadReplica(tp);
if (!clearedReplicaId.isPresent()) {
// If there's no preferred replica to clear, we're
fetching from the leader so handle this error normally
- if (fetchOffset != subscriptions.position(tp).offset) {
+ FetchPosition position = subscriptions.position(tp);
+ if (position != null && fetchOffset != position.offset) {
log.debug("Discarding stale fetch response for
partition {} since the fetched offset {} " +
- "does not match the current offset {}", tp,
fetchOffset, subscriptions.position(tp));
+ "does not match the current offset {}", tp,
fetchOffset, position);
} else {
- handleOffsetOutOfRange(subscriptions.position(tp), tp,
"error response in offset fetch");
+ handleOffsetOutOfRange(position, tp, "error response
in offset fetch");
Review comment:
What should we do here for null position? This can happen if we get OOOR
while we're in the middle of a reset. Maybe we should just log a warning?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
