[wip] Kafka 18376 avoid events

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

@@ -110,6 +110,16 @@ public abstract class AbstractMembershipManager<R extends AbstractResponse> impl
      */
     private final ConsumerMetadata metadata;
 
+    /**
+     * Keeps track of the auto-commit state.
+     *
+     * <p/>
+     *
+     * <em>Note</em>: per its class name, this state is <em>shared</em> with the application thread, so care must be
+     * taken to evaluate how it's used elsewhere when updating related logic.
+     */
+    protected final ThreadSafeAutoCommitState autoCommitState;
+
     /**
      * Logger.
      */
@@ -137,10 +147,17 @@ public abstract class AbstractMembershipManager<R extends AbstractResponse> impl
 
     /**
      * If there is a reconciliation running (triggering commit, callbacks) for the
-     * assignmentReadyToReconcile. This will be true if {@link #maybeReconcile(boolean)} has been triggered
-     * after receiving a heartbeat response, or a metadata update.
+     * assignmentReadyToReconcile. {@link ThreadSafeReconciliationState#isInProgress()} will be true if
+     * {@link #maybeReconcile(boolean)} has been triggered after receiving a heartbeat response, or a metadata update.
+     * Calling code should generally favor {@link #reconciliationInProgress()} for its clarity over direct use of
+     * this state.
+     *
+     * <p/>
+     *
+     * <em>Note</em>: per its class name, this state is <em>shared</em> with the application thread, so care must be
+     * taken to evaluate how it's used elsewhere when updating related logic.
      */
-    private boolean reconciliationInProgress;
+    private final ThreadSafeReconciliationState reconciliationState;
 
     /**
      * True if a reconciliation is in progress and the member rejoined the group since the start
@@ -192,8 +209,6 @@ public abstract class AbstractMembershipManager<R extends AbstractResponse> impl
      */
     private boolean isPollTimerExpired;
 
-    private final boolean autoCommitEnabled;
-
     /**
      * Indicate the operation on consumer group membership that the consumer will perform when leaving the group.
      * The property should remain {@code GroupMembershipOperation.DEFAULT} until the consumer is closing.
@@ -208,7 +223,7 @@ public abstract class AbstractMembershipManager<R extends AbstractResponse> impl
                               Logger log,
                               Time time,
                               RebalanceMetricsManager metricsManager,
-                              boolean autoCommitEnabled) {
+                              ThreadSafeConsumerState threadSafeConsumerState) {
         this.groupId = groupId;
         this.state = MemberState.UNSUBSCRIBED;
         this.subscriptions = subscriptions;
@@ -220,7 +235,8 @@ public abstract class AbstractMembershipManager<R extends AbstractResponse> impl
         this.stateUpdatesListeners = new ArrayList<>();
         this.time = time;
         this.metricsManager = metricsManager;
-        this.autoCommitEnabled = autoCommitEnabled;
+        this.autoCommitState = threadSafeConsumerState.autoCommitState();
+        this.reconciliationState = threadSafeConsumerState.reconciliationState();
     }
 
     /**
@@ -530,7 +546,7 @@ public void transitionToJoining() {
                     "the member is in FATAL state");
             return;
         }
-        if (reconciliationInProgress) {
+        if (reconciliationInProgress()) {
             rejoinedWhileReconciliationInProgress = true;
         }
         resetEpoch();
@@ -830,7 +846,7 @@ public void maybeReconcile(boolean canCommit) {
                     "current assignment.");
             return;
         }
-        if (reconciliationInProgress) {
+        if (reconciliationInProgress()) {
             log.trace("Ignoring reconciliation attempt. Another reconciliation is already in progress. " +
                  "Assignment {} will be handled in the next reconciliation loop.", currentTargetAssignment);
             return;
@@ -850,7 +866,7 @@ public void maybeReconcile(boolean canCommit) {
             return;
         }
 
-        if (autoCommitEnabled && !canCommit) return;
+        if (autoCommitState.isAutoCommitEnabled() && !canCommit) return;
         markReconciliationInProgress();
 
         // Keep copy of assigned TopicPartitions created from the TopicIdPartitions that are
@@ -963,7 +979,7 @@ private void revokeAndAssign(LocalAssignment resolvedAssignment,
                 log.error("Reconciliation failed.", error);
                 markReconciliationCompleted();
             } else {
-                if (reconciliationInProgress && !maybeAbortReconciliation()) {
+                if (reconciliationInProgress() && !maybeAbortReconciliation()) {
                     currentAssignment = resolvedAssignment;
 
                     signalReconciliationCompleting();
@@ -1034,15 +1050,15 @@ protected CompletableFuture<Void> signalPartitionsLost(Set<TopicPartition> parti
      *  Visible for testing.
      */
     void markReconciliationInProgress() {
-        reconciliationInProgress = true;
+        reconciliationState.setInProgress(true);
         rejoinedWhileReconciliationInProgress = false;
     }
 
     /**
      *  Visible for testing.
      */
     void markReconciliationCompleted() {
-        reconciliationInProgress = false;
+        reconciliationState.setInProgress(false);
         rejoinedWhileReconciliationInProgress = false;
     }
 
@@ -1372,7 +1388,7 @@ Map<Uuid, SortedSet<Integer>> topicPartitionsAwaitingReconciliation() {
      * by a call to {@link #maybeReconcile(boolean)}. Visible for testing.
      */
     boolean reconciliationInProgress() {
-        return reconciliationInProgress;
+        return reconciliationState.isInProgress();
     }
 
     /**

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

@@ -320,13 +320,11 @@ private StreamsRebalanceListenerInvoker streamsRebalanceListenerInvoker() {
     private final long retryBackoffMs;
     private final int requestTimeoutMs;
     private final Duration defaultApiTimeoutMs;
-    private final boolean autoCommitEnabled;
+    private final ThreadSafeAsyncConsumerState threadSafeConsumerState;
     private volatile boolean closed = false;
     // Init value is needed to avoid NPE in case of exception raised in the constructor
     private Optional<ClientTelemetryReporter> clientTelemetryReporter = Optional.empty();
 
-    // to keep from repeatedly scanning subscriptions in poll(), cache the result during metadata updates
-    private boolean cachedSubscriptionHasAllFetchPositions;
     private final WakeupTrigger wakeupTrigger = new WakeupTrigger();
     private final OffsetCommitCallbackInvoker offsetCommitCallbackInvoker;
     private final ConsumerRebalanceListenerInvoker rebalanceListenerInvoker;
@@ -388,7 +386,6 @@ public AsyncKafkaConsumer(final ConsumerConfig config,
                 GroupRebalanceConfig.ProtocolType.CONSUMER
             );
             this.clientId = config.getString(CommonClientConfigs.CLIENT_ID_CONFIG);
-            this.autoCommitEnabled = config.getBoolean(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG);
             LogContext logContext = createLogContext(config, groupRebalanceConfig);
             this.backgroundEventQueue = backgroundEventQueue;
             this.log = logContext.logger(getClass());
@@ -430,6 +427,19 @@ public AsyncKafkaConsumer(final ConsumerConfig config,
 
             // This FetchBuffer is shared between the application and network threads.
             this.fetchBuffer = new FetchBuffer(logContext);
+            this.threadSafeConsumerState = new ThreadSafeAsyncConsumerState(
+                ThreadSafeAutoCommitState.fromConfig(
+                    logContext,
+                    config,
+                    time
+                ),
+                logContext,
+                metadata,
+                subscriptions,
+                time,
+                retryBackoffMs,
+                apiVersions
+            );
             final Supplier<NetworkClientDelegate> networkClientDelegateSupplier = NetworkClientDelegate.supplier(time,
                     logContext,
                     metadata,
@@ -440,7 +450,8 @@ public AsyncKafkaConsumer(final ConsumerConfig config,
                     clientTelemetryReporter.map(ClientTelemetryReporter::telemetrySender).orElse(null),
                     backgroundEventHandler,
                     false,
-                    asyncConsumerMetrics
+                    asyncConsumerMetrics,
+                    threadSafeConsumerState
             );
             this.offsetCommitCallbackInvoker = new OffsetCommitCallbackInvoker(interceptors);
             this.groupMetadata.set(initializeGroupMetadata(config, groupRebalanceConfig));
@@ -459,7 +470,8 @@ public AsyncKafkaConsumer(final ConsumerConfig config,
                     metrics,
                     offsetCommitCallbackInvoker,
                     memberStateListener,
-                    streamsRebalanceData
+                    streamsRebalanceData,
+                    threadSafeConsumerState
             );
             final Supplier<ApplicationEventProcessor> applicationEventProcessorSupplier = ApplicationEventProcessor.supplier(logContext,
                     metadata,
@@ -473,7 +485,8 @@ public AsyncKafkaConsumer(final ConsumerConfig config,
                     applicationEventProcessorSupplier,
                     networkClientDelegateSupplier,
                     requestManagersSupplier,
-                    asyncConsumerMetrics
+                    asyncConsumerMetrics,
+                    threadSafeConsumerState
             );
             this.rebalanceListenerInvoker = new ConsumerRebalanceListenerInvoker(
                     logContext,
@@ -532,7 +545,7 @@ public AsyncKafkaConsumer(final ConsumerConfig config,
                        int requestTimeoutMs,
                        int defaultApiTimeoutMs,
                        String groupId,
-                       boolean autoCommitEnabled) {
+                       ThreadSafeAsyncConsumerState threadSafeConsumerState) {
         this.log = logContext.logger(getClass());
         this.subscriptions = subscriptions;
         this.clientId = clientId;
@@ -557,7 +570,7 @@ public AsyncKafkaConsumer(final ConsumerConfig config,
         this.kafkaConsumerMetrics = new KafkaConsumerMetrics(metrics);
         this.asyncConsumerMetrics = new AsyncConsumerMetrics(metrics, CONSUMER_METRIC_GROUP);
         this.clientTelemetryReporter = Optional.empty();
-        this.autoCommitEnabled = autoCommitEnabled;
+        this.threadSafeConsumerState = threadSafeConsumerState;
         this.offsetCommitCallbackInvoker = new OffsetCommitCallbackInvoker(interceptors);
         this.backgroundEventHandler = new BackgroundEventHandler(
             backgroundEventQueue,
@@ -577,7 +590,6 @@ public AsyncKafkaConsumer(final ConsumerConfig config,
         this.log = logContext.logger(getClass());
         this.subscriptions = subscriptions;
         this.clientId = config.getString(ConsumerConfig.CLIENT_ID_CONFIG);
-        this.autoCommitEnabled = config.getBoolean(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG);
         this.fetchBuffer = new FetchBuffer(logContext);
         this.isolationLevel = IsolationLevel.READ_UNCOMMITTED;
         this.time = time;
@@ -623,6 +635,19 @@ public AsyncKafkaConsumer(final ConsumerConfig config,
             new RebalanceCallbackMetricsManager(metrics)
         );
         ApiVersions apiVersions = new ApiVersions();
+        this.threadSafeConsumerState = new ThreadSafeAsyncConsumerState(
+            ThreadSafeAutoCommitState.fromConfig(
+                logContext,
+                config,
+                time
+            ),
+            logContext,
+            metadata,
+            subscriptions,
+            time,
+            retryBackoffMs,
+            apiVersions
+        );
         Supplier<NetworkClientDelegate> networkClientDelegateSupplier = () -> new NetworkClientDelegate(
             time,
             config,
@@ -631,7 +656,8 @@ public AsyncKafkaConsumer(final ConsumerConfig config,
             metadata,
             backgroundEventHandler,
             false,
-            asyncConsumerMetrics
+            asyncConsumerMetrics,
+            threadSafeConsumerState
         );
         this.offsetCommitCallbackInvoker = new OffsetCommitCallbackInvoker(interceptors);
         Supplier<RequestManagers> requestManagersSupplier = RequestManagers.supplier(
@@ -650,7 +676,8 @@ public AsyncKafkaConsumer(final ConsumerConfig config,
             metrics,
             offsetCommitCallbackInvoker,
             memberStateListener,
-            Optional.empty()
+            Optional.empty(),
+            threadSafeConsumerState
         );
         Supplier<ApplicationEventProcessor> applicationEventProcessorSupplier = ApplicationEventProcessor.supplier(
                 logContext,
@@ -665,7 +692,8 @@ public AsyncKafkaConsumer(final ConsumerConfig config,
                 applicationEventProcessorSupplier,
                 networkClientDelegateSupplier,
                 requestManagersSupplier,
-                asyncConsumerMetrics);
+                asyncConsumerMetrics,
+                threadSafeConsumerState);
         this.streamsRebalanceListenerInvoker = Optional.empty();
         this.backgroundEventProcessor = new BackgroundEventProcessor();
         this.backgroundEventReaper = new CompletableEventReaper(logContext);
@@ -682,7 +710,8 @@ ApplicationEventHandler build(
             final Supplier<ApplicationEventProcessor> applicationEventProcessorSupplier,
             final Supplier<NetworkClientDelegate> networkClientDelegateSupplier,
             final Supplier<RequestManagers> requestManagersSupplier,
-            final AsyncConsumerMetrics asyncConsumerMetrics
+            final AsyncConsumerMetrics asyncConsumerMetrics,
+            final ThreadSafeConsumerState threadSafeConsumerState
         );
 
     }
@@ -833,14 +862,7 @@ public ConsumerRecords<K, V> poll(final Duration timeout) {
             }
 
             do {
-                PollEvent event = new PollEvent(timer.currentTimeMs());
-                // Make sure to let the background thread know that we are still polling.
-                // This will trigger async auto-commits of consumed positions when hitting
-                // the interval time or reconciling new assignments
-                applicationEventHandler.add(event);
-                // Wait for reconciliation and auto-commit to be triggered, to ensure all commit requests
-                // retrieve the positions to commit before proceeding with fetching new records
-                ConsumerUtils.getResult(event.reconcileAndAutoCommit(), defaultApiTimeoutMs.toMillis());
+                sendPollEvent(timer);
 
                 // We must not allow wake-ups between polling for fetches and returning the records.
                 // If the polled fetches are not empty the consumed position has already been updated in the polling
@@ -876,6 +898,29 @@ public ConsumerRecords<K, V> poll(final Duration timeout) {
         }
     }
 
+    private void sendPollEvent(Timer timer) {
+        long currentTimeMs = timer.currentTimeMs();
+
+        // Make sure to let the background thread know that we are still polling.
+        PollEvent event = new PollEvent(currentTimeMs);
+
+        if (threadSafeConsumerState.canSkipWaitingOnPoll(currentTimeMs)) {
+            // This will *not* trigger async auto-commits of consumed positions as the shared Timer for
+            // auto-commit interval will not change between the application thread and the network thread. This
+            // is true of the reconciliation state. The state will not change between the SharedConsumerState
+            // check above and the processing of the PollEvent.
+            applicationEventHandler.add(event);
+        } else {
+            // This will trigger async auto-commits of consumed positions when hitting
+            // the interval time or reconciling new assignments
+            applicationEventHandler.add(event);
+
+            // Wait for reconciliation and auto-commit to be triggered, to ensure all commit requests
+            // retrieve the positions to commit before proceeding with fetching new records
+            ConsumerUtils.getResult(event.reconcileAndAutoCommit(), defaultApiTimeoutMs.toMillis());
+        }
+    }
+
     /**
      * Commit offsets returned on the last {@link #poll(Duration) poll()} for all the subscribed list of topics and
      * partitions.
@@ -1488,7 +1533,7 @@ private void autoCommitOnClose(final Timer timer) {
         if (groupMetadata.get().isEmpty() || applicationEventHandler == null)
             return;
 
-        if (autoCommitEnabled)
+        if (threadSafeConsumerState.autoCommitState().isAutoCommitEnabled())
             commitSyncAllConsumed(timer);
 
         applicationEventHandler.add(new CommitOnCloseEvent());
@@ -1777,9 +1822,9 @@ private Fetch<K, V> pollForFetches(Timer timer) {
         // We do not want to be stuck blocking in poll if we are missing some positions
         // since the offset lookup may be backing off after a failure
 
-        // NOTE: the use of cachedSubscriptionHasAllFetchPositions means we MUST call
+        // NOTE: for hasAllFetchPositions to return the correct answer, we MUST call
         // updateAssignmentMetadataIfNeeded before this method.
-        if (!cachedSubscriptionHasAllFetchPositions && pollTimeout > retryBackoffMs) {
+        if (!subscriptions.hasAllFetchPositions() && pollTimeout > retryBackoffMs) {
             pollTimeout = retryBackoffMs;
         }
 
@@ -1834,11 +1879,21 @@ private Fetch<K, V> collectFetch() {
      *                                       defined
      */
     private boolean updateFetchPositions(final Timer timer) {
-        cachedSubscriptionHasAllFetchPositions = false;
+        // Fetch position validation is in the hot path for poll() and the cost of thread interaction for
+        // event processing is *very* heavy, CPU-wise. In a stable system, the positions are valid; having the
+        // network thread check the validity yields the same answer 99%+ of the time. But calling the
+        // network thread to determine that is very expensive.
+        //
+        // Instead, let the *application thread* determine if any partitions need their positions updated. If not,
+        // the application thread can skip sending an event to the network thread that will simply end up coming
+        // to the same conclusion, albeit much slower.
+        if (threadSafeConsumerState.canSkipUpdateFetchPositions())
+            return true;
+
         try {
             CheckAndUpdatePositionsEvent checkAndUpdatePositionsEvent = new CheckAndUpdatePositionsEvent(calculateDeadlineMs(timer));
             wakeupTrigger.setActiveTask(checkAndUpdatePositionsEvent.future());
-            cachedSubscriptionHasAllFetchPositions = applicationEventHandler.addAndGet(checkAndUpdatePositionsEvent);
+            applicationEventHandler.addAndGet(checkAndUpdatePositionsEvent);
         } catch (TimeoutException e) {
             return false;
         } finally {