KAFKA-15942: Implement ConsumerInterceptors in the async consumer

clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java

@@ -282,7 +282,6 @@ private void process(final ConsumerRebalanceListenerCallbackNeededEvent event) {
     // to keep from repeatedly scanning subscriptions in poll(), cache the result during metadata updates
     private boolean cachedSubscriptionHasAllFetchPositions;
     private final WakeupTrigger wakeupTrigger = new WakeupTrigger();
-    private boolean isFenced = false;
     private final OffsetCommitCallbackInvoker offsetCommitCallbackInvoker;
 
     // currentThread holds the threadId of the current thread accessing the AsyncKafkaConsumer
@@ -789,7 +788,6 @@ CompletableFuture<Void> commit(final Map<TopicPartition, OffsetAndMetadata> offs
                                    final boolean isWakeupable,
                                    final Optional<Long> retryTimeoutMs) {
         maybeInvokeCommitCallbacks();
-        maybeThrowFencedInstanceException();
         maybeThrowInvalidGroupIdException();
 
         log.debug("Committing offsets: {}", offsets);
@@ -1643,7 +1641,6 @@ private void updateLastSeenEpochIfNewer(TopicPartition topicPartition, OffsetAnd
     @Override
     public boolean updateAssignmentMetadataIfNeeded(Timer timer) {
         maybeInvokeCommitCallbacks();
-        maybeThrowFencedInstanceException();
         backgroundEventProcessor.process();
 
         // Keeping this updateAssignmentMetadataIfNeeded wrapping up the updateFetchPositions as
@@ -1897,8 +1894,9 @@ public KafkaConsumerMetrics kafkaConsumerMetrics() {
         return kafkaConsumerMetrics;
     }
 
-    private void maybeThrowFencedInstanceException() {
-        if (isFenced) {
+    private void maybeInvokeCommitCallbacks() {
+        offsetCommitCallbackInvoker.executeCallbacks();
+        if (offsetCommitCallbackInvoker.hasFencedException()) {
             String groupInstanceId = "unknown";
             if (!groupMetadata.isPresent()) {
                 log.error("No group metadata found although a group ID was provided. This is a bug!");
@@ -1911,12 +1909,6 @@ private void maybeThrowFencedInstanceException() {
         }
     }
 
-    private void maybeInvokeCommitCallbacks() {
-        if (offsetCommitCallbackInvoker.executeCallbacks()) {
-            isFenced = true;
-        }
-    }
-
     // Visible for testing
     SubscriptionState subscriptions() {
         return subscriptions;

clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java

@@ -97,10 +97,17 @@ public CommitRequestManager(
             final OffsetCommitCallbackInvoker offsetCommitCallbackInvoker,
             final String groupId,
             final Optional<String> groupInstanceId) {
-        this(time, logContext, subscriptions, config, coordinatorRequestManager,
-                offsetCommitCallbackInvoker, groupId,
-                groupInstanceId, config.getLong(ConsumerConfig.RETRY_BACKOFF_MS_CONFIG),
-                config.getLong(ConsumerConfig.RETRY_BACKOFF_MAX_MS_CONFIG), OptionalDouble.empty());
+        this(time,
+            logContext,
+            subscriptions,
+            config,
+            coordinatorRequestManager,
+            offsetCommitCallbackInvoker,
+            groupId,
+            groupInstanceId,
+            config.getLong(ConsumerConfig.RETRY_BACKOFF_MS_CONFIG),
+            config.getLong(ConsumerConfig.RETRY_BACKOFF_MAX_MS_CONFIG),
+            OptionalDouble.empty());
     }
 
     // Visible for testing

clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetCommitCallbackInvoker.java

@@ -33,6 +33,7 @@
  */
 public class OffsetCommitCallbackInvoker {
     private final ConsumerInterceptors<?, ?> interceptors;
+    private boolean hasFencedException = false;
 
     OffsetCommitCallbackInvoker(ConsumerInterceptors<?, ?> interceptors) {
         this.interceptors = interceptors;
@@ -57,23 +58,22 @@ public void submitUserCallback(final OffsetCommitCallback callback,
         callbackQueue.add(new OffsetCommitCallbackTask(callback, offsets, exception));
     }
 
-    /**
-     * @return true if an offset commit was fenced.
-     */
-    public boolean executeCallbacks() {
-        boolean isFenced = false;
+    public void executeCallbacks() {
         while (!callbackQueue.isEmpty()) {
             OffsetCommitCallbackTask task = callbackQueue.poll();
             if (task != null) {
 
                 if (task.exception instanceof FencedInstanceIdException)
-                    isFenced = true;
+                    hasFencedException = true;
 
                 task.callback.onComplete(task.offsets, task.exception);
 
             }
         }
-        return isFenced;
+    }
+
+    public boolean hasFencedException() {
+        return hasFencedException;
     }
 
     private static class OffsetCommitCallbackTask {

clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java

@@ -849,6 +849,22 @@ public void testWakeupCommitted() {
         assertNull(consumer.wakeupTrigger().getPendingTask());
     }
 
+    @Test
+    public void testInterceptorAutoCommitOnClose() {
+        Properties props = requiredConsumerPropertiesAndGroupId("test-id");
+        props.setProperty(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, MockConsumerInterceptor.class.getName());
+        props.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
+
+        consumer = newConsumer(props);
+        assertEquals(1, MockConsumerInterceptor.INIT_COUNT.get());
+        completeCommitApplicationEventSuccessfully();
+
+        consumer.close(Duration.ZERO);
+
+        assertEquals(1, MockConsumerInterceptor.ON_COMMIT_COUNT.get());
+        assertEquals(1, MockConsumerInterceptor.CLOSE_COUNT.get());
+    }
+
     @Test
     public void testInterceptorCommitSync() {
         Properties props = requiredConsumerPropertiesAndGroupId("test-id");

core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala

@@ -1336,8 +1336,9 @@ class PlaintextConsumerTest extends BaseConsumerTest {
     MockProducerInterceptor.resetCounters()
   }
 
+  // This is disabled for the the consumer group until KAFKA-16155 is resolved.
   @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
-  @MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
+  @MethodSource(Array("getTestQuorumAndGroupProtocolParametersClassicGroupProtocolOnly"))
   def testAutoCommitIntercept(quorum: String, groupProtocol: String): Unit = {
     val topic2 = "topic2"
     createTopic(topic2, 2, brokerCount)