@Override
            protected RequestManagers create() {
                long retryBackoffMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MS_CONFIG);
                long retryBackoffMaxMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MAX_MS_CONFIG);
                FetchConfig fetchConfig = new FetchConfig(config);

                CoordinatorRequestManager coordinator = new CoordinatorRequestManager(
                        logContext,
                        retryBackoffMs,
                        retryBackoffMaxMs,
                        backgroundEventHandler,
                        groupRebalanceConfig.groupId);
                ShareMembershipManager shareMembershipManager = new ShareMembershipManager(
                        logContext,
                        groupRebalanceConfig.groupId,
                        null,
                        subscriptions,
                        metadata,
                        clientTelemetryReporter,
                        time,
                        metrics);
                ShareHeartbeatRequestManager shareHeartbeatRequestManager = new ShareHeartbeatRequestManager(
                        logContext,
                        time,
                        config,
                        coordinator,
                        subscriptions,
                        shareMembershipManager,
                        backgroundEventHandler,
                        metrics);
                ShareConsumeRequestManager shareConsumeRequestManager = new ShareConsumeRequestManager(
                        time,
                        logContext,
                        groupRebalanceConfig.groupId,
                        metadata,
                        subscriptions,
                        fetchConfig,
                        fetchBuffer,
                        backgroundEventHandler,
                        shareFetchMetricsManager,
                        retryBackoffMs,
                        retryBackoffMaxMs);
                shareMembershipManager.registerStateListener(shareConsumeRequestManager);

                return new RequestManagers(
                        logContext,
                        shareConsumeRequestManager,
                        Optional.of(coordinator),
                        Optional.of(shareHeartbeatRequestManager),
                        Optional.of(shareMembershipManager)
                );
            }
Creates a {@link Supplier} for deferred creation during invocation by
{@link ShareConsumerImpl}.
clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestManagers.java