@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