mumrah commented on a change in pull request #8841:
URL: https://github.com/apache/kafka/pull/8841#discussion_r441933113
##########
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:
I'm inclined to keep the same behavior as we previously did and just
warn (not that we expect this case to get hit anymore)
----------------------------------------------------------------
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
