philipnee commented on code in PR #14680:
URL: https://github.com/apache/kafka/pull/14680#discussion_r1388387504
##########
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() {
+ 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 {@link OffsetCommitCallback}. This is
+ * achieved by having the background thread to register a {@link
OffsetCommitCallbackTask} to the invoker upon the
+ * future completion, and execute the callbacks when user
polls/commits/closes the consumer.
+ */
+ private class OffsetCommitCallbackInvoker {
+ // Thread-safe queue to store callbacks
+ private final BlockingQueue<OffsetCommitCallbackTask> callbackQueue =
new LinkedBlockingQueue<>();
+
+ public void submit(final OffsetCommitCallbackTask callback) {
+ try {
+ callbackQueue.offer(callback);
+ } catch (Exception e) {
+ log.error("Failed to enqueue OffsetCommitCallback", e);
+ }
+ }
+
+ public void executeCallbacks() {
+ LinkedList<OffsetCommitCallbackTask> callbacks = new
LinkedList<>();
Review Comment:
Thanks, I checked the current implementation, we don't need to do this.
--
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
