Repository: apache/kafka
File: clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java (Java)

Code before:
        }
    }

    private void process(final FetchCommittedOffsetsEvent event) {
        if (requestManagers.commitRequestManager.isEmpty()) {
            event.future().completeExceptionally(new KafkaException("Unable to fetch committed " +
                    "offset because the CommitRequestManager is not available. Check if group.id was set correctly"));
            return;
        }
        CommitRequestManager manager = requestManagers.commitRequestManager.get();
        CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future = manager.fetchOffsets(event.partitions(), event.deadlineMs());
        future.whenComplete(complete(event.future()));
    }

    /**
     * Commit all consumed if auto-commit is enabled. Note this will trigger an async commit,
     * that will not be retried if the commit request fails.
     */
    private void process(final AssignmentChangeEvent event) {
        if (requestManagers.commitRequestManager.isPresent()) {
            CommitRequestManager manager = requestManagers.commitRequestManager.get();
            manager.updateTimerAndMaybeCommit(event.currentTimeMs());
        }

        if (log.isInfoEnabled())
            log.info("Assigned to partition(s): {}", Utils.topicPartitionString(event.partitions()));

        try {
            if (subscriptions.assignFromUser(new HashSet<>(event.partitions())))
                metadata.requestUpdateForNewTopics();

            event.future().complete(null);
        } catch (Exception e) {
            event.future().completeExceptionally(e);
        }
    }

    /**
     * Handles ListOffsetsEvent by fetching the offsets for the given partitions and timestamps.
     */
    private void process(final ListOffsetsEvent event) {
        final CompletableFuture<Map<TopicPartition, OffsetAndTimestampInternal>> future =
            requestManagers.offsetsRequestManager.fetchOffsets(event.timestampsToSearch(), event.requireTimestamps());
        future.whenComplete(complete(event.future()));
    }

    /**
     * Process event that indicates that the subscription topics changed. This will make the
     * consumer join the group if it is not part of it yet, or send the updated subscription if
     * it is already a member on the next poll.
     */

Review comment (suggestion):
ok with the change, even though this one is not on any hot-path I expect, right? (just API calls to consumer.assign that do a single trip to the background, no requests). 

Code after:
    }

    private void process(final FetchCommittedOffsetsEvent event) {
        if (requestManagers.commitRequestManager.isEmpty()) {
            event.future().completeExceptionally(new KafkaException("Unable to fetch committed " +
                    "offset because the CommitRequestManager is not available. Check if group.id was set correctly"));
            return;
        }
        CommitRequestManager manager = requestManagers.commitRequestManager.get();
        CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future = manager.fetchOffsets(event.partitions(), event.deadlineMs());
        future.whenComplete(complete(event.future()));
    }

    /**
     * Commit all consumed if auto-commit is enabled. Note this will trigger an async commit,
     * that will not be retried if the commit request fails.
     */
    private void process(final AssignmentChangeEvent event) {
        if (requestManagers.commitRequestManager.isPresent()) {
            CommitRequestManager manager = requestManagers.commitRequestManager.get();
            manager.updateTimerAndMaybeCommit(event.currentTimeMs());
        }

        log.info("Assigned to partition(s): {}", event.partitions());
        try {
            if (subscriptions.assignFromUser(new HashSet<>(event.partitions())))
                metadata.requestUpdateForNewTopics();

            event.future().complete(null);
        } catch (Exception e) {
            event.future().completeExceptionally(e);
        }
    }

    /**
     * Handles ListOffsetsEvent by fetching the offsets for the given partitions and timestamps.
     */
    private void process(final ListOffsetsEvent event) {
        final CompletableFuture<Map<TopicPartition, OffsetAndTimestampInternal>> future =
            requestManagers.offsetsRequestManager.fetchOffsets(event.timestampsToSearch(), event.requireTimestamps());
        future.whenComplete(complete(event.future()));
    }

    /**
     * Process event that indicates that the subscription topics changed. This will make the
     * consumer join the group if it is not part of it yet, or send the updated subscription if
     * it is already a member on the next poll.
     */
    private void process(final TopicSubscriptionChangeEvent event) {
        if (requestManagers.consumerHeartbeatRequestManager.isPresent()) {
            try {