philipnee commented on code in PR #14680:
URL: https://github.com/apache/kafka/pull/14680#discussion_r1388374351
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumer.java:
##########
@@ -1010,15 +1031,9 @@ private void updateLastSeenEpochIfNewer(TopicPartition
topicPartition, OffsetAnd
offsetAndMetadata.leaderEpoch().ifPresent(epoch ->
metadata.updateLastSeenEpochIfNewer(topicPartition, epoch));
}
- private class DefaultOffsetCommitCallback implements OffsetCommitCallback {
- @Override
- public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets,
Exception exception) {
- if (exception != null)
- log.error("Offset commit with offsets {} failed", offsets,
exception);
- }
- }
-
boolean updateAssignmentMetadataIfNeeded(Timer timer) {
+ maybeInvokeCallbacks();
Review Comment:
Hi, to your first question, it was an oversight, so i fixed it. to your
second question: I moved it down for a couple of reasons, one is we want the
callback invocation to be included in the poll timer, and two, i wanted to
consolidate the logic with the backgroundEventProcess.poll to make its behavior
similar to the current consumer. The current consumer poll
`ConsumerCoordinator` class in the `updateAssignmentMetadataIfNeeded`.
the `ConsumerCoordinator#poll` actually performs the callback invocation
first, which, first checks the fencedInstaceId. See the code below:
```
public boolean poll(Timer timer, boolean waitForJoinGroup) {
maybeUpdateSubscriptionMetadata();
invokeCompletedOffsetCommitCallbacks();
...
}
```
and
```
void invokeCompletedOffsetCommitCallbacks() {
if (asyncCommitFenced.get()) {
throw new FencedInstanceIdException("Get fenced exception for
group.instance.id "
+ rebalanceConfig.groupInstanceId.orElse("unset_instance_id")
+ ", current member.id is " + memberId());
}
while (true) {
OffsetCommitCompletion completion =
completedOffsetCommits.poll();
if (completion == null) {
break;
}
completion.invoke();
}
}
```
--
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
