@SuppressWarnings({"checkstyle:ParameterNumber"})
    public static Supplier<RequestManagers> supplier(final Time time,
                                                     final LogContext logContext,
                                                     final BackgroundEventHandler backgroundEventHandler,
                                                     final ConsumerMetadata metadata,
                                                     final SubscriptionState subscriptions,
                                                     final FetchBuffer fetchBuffer,
                                                     final ConsumerConfig config,
                                                     final GroupRebalanceConfig groupRebalanceConfig,
                                                     final ApiVersions apiVersions,
                                                     final FetchMetricsManager fetchMetricsManager,
                                                     final Supplier<NetworkClientDelegate> networkClientDelegateSupplier,
                                                     final Optional<ClientTelemetryReporter> clientTelemetryReporter,
                                                     final Metrics metrics,
                                                     final OffsetCommitCallbackInvoker offsetCommitCallbackInvoker,
                                                     final MemberStateListener applicationThreadMemberStateListener
                                                     ) {
        return new CachedSupplier<RequestManagers>() {
            @Override
            protected RequestManagers create() {
                final NetworkClientDelegate networkClientDelegate = networkClientDelegateSupplier.get();
                final FetchConfig fetchConfig = new FetchConfig(config);
                long retryBackoffMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MS_CONFIG);
                long retryBackoffMaxMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MAX_MS_CONFIG);
                final int requestTimeoutMs = config.getInt(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG);
                final OffsetsRequestManager listOffsets = new OffsetsRequestManager(subscriptions,
                        metadata,
                        fetchConfig.isolationLevel,
                        time,
                        retryBackoffMs,
                        requestTimeoutMs,
                        apiVersions,
                        networkClientDelegate,
                        backgroundEventHandler,
                        logContext);
                final FetchRequestManager fetch = new FetchRequestManager(logContext,
                        time,
                        metadata,
                        subscriptions,
                        fetchConfig,
                        fetchBuffer,
                        fetchMetricsManager,
                        networkClientDelegate,
                        apiVersions);
                final TopicMetadataRequestManager topic = new TopicMetadataRequestManager(
                        logContext,
                        time,
                        config);
                HeartbeatRequestManager heartbeatRequestManager = null;
                MembershipManager membershipManager = null;
                CoordinatorRequestManager coordinator = null;
                CommitRequestManager commit = null;

                if (groupRebalanceConfig != null && groupRebalanceConfig.groupId != null) {
                    Optional<String> serverAssignor = Optional.ofNullable(config.getString(ConsumerConfig.GROUP_REMOTE_ASSIGNOR_CONFIG));
                    coordinator = new CoordinatorRequestManager(
                            logContext,
                            retryBackoffMs,
                            retryBackoffMaxMs,
                            backgroundEventHandler,
                            groupRebalanceConfig.groupId);
                    commit = new CommitRequestManager(
                            time,
                            logContext,
                            subscriptions,
                            config,
                            coordinator,
                            offsetCommitCallbackInvoker,
                            groupRebalanceConfig.groupId,
                            groupRebalanceConfig.groupInstanceId,
                            metrics);
                    membershipManager = new MembershipManagerImpl(
                            groupRebalanceConfig.groupId,
                            groupRebalanceConfig.groupInstanceId,
                            groupRebalanceConfig.rebalanceTimeoutMs,
                            serverAssignor,
                            subscriptions,
                            commit,
                            metadata,
                            logContext,
                            clientTelemetryReporter,
                            backgroundEventHandler,
                            time,
                            metrics);
                    membershipManager.registerStateListener(commit);
                    membershipManager.registerStateListener(applicationThreadMemberStateListener);
                    heartbeatRequestManager = new HeartbeatRequestManager(
                            logContext,
                            time,
                            config,
                            coordinator,
                            subscriptions,
                            membershipManager,
                            backgroundEventHandler,
                            metrics);
                }

                return new RequestManagers(
                        logContext,
                        listOffsets,
                        topic,
                        fetch,
                        Optional.ofNullable(coordinator),
                        Optional.ofNullable(commit),
                        Optional.ofNullable(heartbeatRequestManager),
                        Optional.ofNullable(membershipManager)
                );
            }
        };
    }
Creates a {@link Supplier} for deferred creation during invocation by
{@link AsyncKafkaConsumer}.
clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestManagers.java