philipnee commented on code in PR #14680:
URL: https://github.com/apache/kafka/pull/14680#discussion_r1380360543
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumer.java:
##########
@@ -1096,4 +1114,47 @@ private void subscribeInternal(Collection<String>
topics, Optional<ConsumerRebal
}
}
+ private void maybeThrowFencedInstanceException() {
+ if (fencedInstance.get()) {
+ throw new FencedInstanceIdException("Get fenced exception for
group.instance.id " +
+ groupInstanceId.orElse("null"));
+ }
+ }
+
+ // Visible for testing
+ void maybeInvokeCallbacks() {
+ if (callbacks() > 0) {
+ invoker.executeCallbacks();
+ }
+ }
+
+ // Visible for testing
+ int callbacks() {
+ return invoker.callbackQueue.size();
+ }
+
+ /**
+ * Utility class that helps the application thread to invoke user
registered callbacks such as
+ * {@link OffsetCommitCallback}. This is achieved by having the
background thread to register a runnable to the
+ * invoker upon the future completion, and execute the callbacks when user
polls the consumer.
+ */
+ private static class CallbackInvoker {
+ // Thread-safe queue to store callbacks
+ private final BlockingQueue<Runnable> callbackQueue = new
LinkedBlockingQueue<>();
Review Comment:
Possible. However, the OffsetCommitCallback was not registered to the
background thread, that why I added an invoker.
--
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
