@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;
                ConsumerMembershipManager 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 ConsumerMembershipManager(
                            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)
                );
            }
        };
    }
@Test
    public void testMemberStateListenerRegistered() {

        final MemberStateListener listener = (memberEpoch, memberId) -> { };

        final Properties properties = requiredConsumerConfig();
        properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "consumerGroup");
        final ConsumerConfig config = new ConsumerConfig(properties);
        final GroupRebalanceConfig groupRebalanceConfig = new GroupRebalanceConfig(
            config,
            GroupRebalanceConfig.ProtocolType.CONSUMER
        );
        final RequestManagers requestManagers = RequestManagers.supplier(
            new MockTime(),
            new LogContext(),
            mock(BackgroundEventHandler.class),
            mock(ConsumerMetadata.class),
            mock(SubscriptionState.class),
            mock(FetchBuffer.class),
            config,
            groupRebalanceConfig,
            mock(ApiVersions.class),
            mock(FetchMetricsManager.class),
            () -> mock(NetworkClientDelegate.class),
            Optional.empty(),
            new Metrics(),
            mock(OffsetCommitCallbackInvoker.class),
            listener
        ).get();
        requestManagers.consumerMembershipManager.ifPresent(
            membershipManager -> assertTrue(membershipManager.stateListeners().contains(listener))
        );
    }