philipnee commented on code in PR #14680:
URL: https://github.com/apache/kafka/pull/14680#discussion_r1388377068
##########
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:
The `backgroundEventProcessor.process` is very similar to
`ConsumerCoordinator#poll`; however, because of the implementation difference,
the two implementations will differ slightly. The change here is to ensure the
two behave as similar as possible.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumer.java:
##########
@@ -1096,4 +1111,76 @@ private void subscribeInternal(Collection<String>
topics, Optional<ConsumerRebal
}
}
-}
\ No newline at end of file
+ private void maybeThrowFencedInstanceException() {
+ if (isFenced) {
+ throw new FencedInstanceIdException("Get fenced exception for
group.instance.id " +
+ groupInstanceId.orElse("null"));
+ }
+ }
+
+ // Visible for testing
+ void maybeInvokeCallbacks() {
Review Comment:
great!
--
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
