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

Code before:

                return new RequestManagers(
                        logContext,
                        listOffsets,
                        topic,
                        fetch,
                        Optional.ofNullable(coordinator),
                        Optional.ofNullable(commitRequestManager),
                        Optional.ofNullable(heartbeatRequestManager),
                        Optional.ofNullable(membershipManager),
                        Optional.ofNullable(streamsGroupHeartbeatRequestManager),
                        Optional.ofNullable(streamsMembershipManager)
                );
            }
        };
    }

    /**
     * Creates a {@link Supplier} for deferred creation during invocation by
     * {@link ShareConsumerImpl}.
     */
    @SuppressWarnings({"checkstyle:ParameterNumber"})
    public static Supplier<RequestManagers> supplier(final Time time,
                                                     final LogContext logContext,
                                                     final BackgroundEventHandler backgroundEventHandler,
                                                     final Metadata metadata,
                                                     final SubscriptionState subscriptions,
                                                     final ShareFetchBuffer fetchBuffer,
                                                     final ConsumerConfig config,
                                                     final GroupRebalanceConfig groupRebalanceConfig,
                                                     final ShareFetchMetricsManager shareFetchMetricsManager,
                                                     final Optional<ClientTelemetryReporter> clientTelemetryReporter,
                                                     final Metrics metrics
    ) {
        return new CachedSupplier<>() {
            @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,
                        groupRebalanceConfig.groupId);
                ShareMembershipManager shareMembershipManager = new ShareMembershipManager(
                        logContext,
                        groupRebalanceConfig.groupId,
                        groupRebalanceConfig.rackId.orElse(null),
                        subscriptions,

Review comment (suggestion):
this `supplier` is only used by the `ShareConsumer`, should we pass `ShareConsumerMetadata` here instead? (consistent with how the other `supplier` used by the regular consumer takes a `ConsumerMetadata`)

Code after:
                return new RequestManagers(
                        logContext,
                        listOffsets,
                        topic,
                        fetch,
                        Optional.ofNullable(coordinator),
                        Optional.ofNullable(commitRequestManager),
                        Optional.ofNullable(heartbeatRequestManager),
                        Optional.ofNullable(membershipManager),
                        Optional.ofNullable(streamsGroupHeartbeatRequestManager),
                        Optional.ofNullable(streamsMembershipManager)
                );
            }
        };
    }

    /**
     * Creates a {@link Supplier} for deferred creation during invocation by
     * {@link ShareConsumerImpl}.
     */
    @SuppressWarnings({"checkstyle:ParameterNumber"})
    public static Supplier<RequestManagers> supplier(final Time time,
                                                     final LogContext logContext,
                                                     final BackgroundEventHandler backgroundEventHandler,
                                                     final ShareConsumerMetadata metadata,
                                                     final SubscriptionState subscriptions,
                                                     final ShareFetchBuffer fetchBuffer,
                                                     final ConsumerConfig config,
                                                     final GroupRebalanceConfig groupRebalanceConfig,
                                                     final ShareFetchMetricsManager shareFetchMetricsManager,
                                                     final Optional<ClientTelemetryReporter> clientTelemetryReporter,
                                                     final Metrics metrics
    ) {
        return new CachedSupplier<>() {
            @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,
                        groupRebalanceConfig.groupId);
                ShareMembershipManager shareMembershipManager = new ShareMembershipManager(
                        logContext,
                        groupRebalanceConfig.groupId,
                        groupRebalanceConfig.rackId.orElse(null),
                        subscriptions,
                        metadata,