Repository: apache/kafka
File: group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java (Java)

Code before:
     * The configured topology including resolved regular expressions.
     */
    private final TimelineObject<Optional<ConfiguredTopology>> configuredTopology;

    /**
     * The metadata refresh deadline. It consists of a timestamp in milliseconds together with the group epoch at the time of setting it.
     * The metadata refresh time is considered as a soft state (read that it is not stored in a timeline data structure). It is like this
     * because it is not persisted to the log. The group epoch is here to ensure that the metadata refresh deadline is invalidated if the
     * group epoch does not correspond to the current group epoch. This can happen if the metadata refresh deadline is updated after having
     * refreshed the metadata but the write operation failed. In this case, the time is not automatically rolled back.
     */
    private DeadlineAndEpoch metadataRefreshDeadline = DeadlineAndEpoch.EMPTY;

    /**
     * Keeps a member ID that requested a shutdown for this group.
     * This has no direct effect inside the group coordinator, but is propagated to old and new members of the group.
     * It is cleared once the group is empty.
     * This is not persisted in the log, as the shutdown request is best-effort.
     */
    private Optional<String> shutdownRequestMemberId = Optional.empty();

    /**
     * The current epoch for endpoint information, this is used to determine when to send
     * updated endpoint information to members of the group.
     */
    private int endpointInformationEpoch = -1;

    public StreamsGroup(
        LogContext logContext,
        SnapshotRegistry snapshotRegistry,
        String groupId,
        GroupCoordinatorMetricsShard metrics
    ) {
        this.log = logContext.logger(StreamsGroup.class);
        this.logContext = logContext;
        this.snapshotRegistry = Objects.requireNonNull(snapshotRegistry);
        this.groupId = Objects.requireNonNull(groupId);
        this.state = new TimelineObject<>(snapshotRegistry, EMPTY);
        this.groupEpoch = new TimelineInteger(snapshotRegistry);
        this.members = new TimelineHashMap<>(snapshotRegistry, 0);
        this.staticMembers = new TimelineHashMap<>(snapshotRegistry, 0);
        this.partitionMetadata = new TimelineHashMap<>(snapshotRegistry, 0);
        this.metadataHash = new TimelineLong(snapshotRegistry);
        this.targetAssignmentEpoch = new TimelineInteger(snapshotRegistry);
        this.targetAssignment = new TimelineHashMap<>(snapshotRegistry, 0);
        this.currentActiveTaskToProcessId = new TimelineHashMap<>(snapshotRegistry, 0);
        this.currentStandbyTaskToProcessIds = new TimelineHashMap<>(snapshotRegistry, 0);
        this.currentWarmupTaskToProcessIds = new TimelineHashMap<>(snapshotRegistry, 0);
        this.metrics = Objects.requireNonNull(metrics);
        this.topology = new TimelineObject<>(snapshotRegistry, Optional.empty());
        this.configuredTopology = new TimelineObject<>(snapshotRegistry, Optional.empty());

Review comment (suggestion):
The removal of the GroupCoordinatorMetricsShard constructor parameter is a breaking change to this public constructor. If StreamsGroup is intended for external (non-internal) use, consider a deprecation cycle or documenting the change in release notes; otherwise, mark the class or constructor clearly as internal to avoid unintended API breakage.

Code after:
     * The metadata refresh time is considered as a soft state (read that it is not stored in a timeline data structure). It is like this
     * because it is not persisted to the log. The group epoch is here to ensure that the metadata refresh deadline is invalidated if the
     * group epoch does not correspond to the current group epoch. This can happen if the metadata refresh deadline is updated after having
     * refreshed the metadata but the write operation failed. In this case, the time is not automatically rolled back.
     */
    private DeadlineAndEpoch metadataRefreshDeadline = DeadlineAndEpoch.EMPTY;

    /**
     * Keeps a member ID that requested a shutdown for this group.
     * This has no direct effect inside the group coordinator, but is propagated to old and new members of the group.
     * It is cleared once the group is empty.
     * This is not persisted in the log, as the shutdown request is best-effort.
     */
    private Optional<String> shutdownRequestMemberId = Optional.empty();

    /**
     * The current epoch for endpoint information, this is used to determine when to send
     * updated endpoint information to members of the group.
     */
    private int endpointInformationEpoch = -1;

    public StreamsGroup(
        LogContext logContext,
        SnapshotRegistry snapshotRegistry,
        String groupId
    ) {
        this.log = logContext.logger(StreamsGroup.class);
        this.logContext = logContext;
        this.snapshotRegistry = Objects.requireNonNull(snapshotRegistry);
        this.groupId = Objects.requireNonNull(groupId);
        this.state = new TimelineObject<>(snapshotRegistry, EMPTY);
        this.groupEpoch = new TimelineInteger(snapshotRegistry);
        this.members = new TimelineHashMap<>(snapshotRegistry, 0);
        this.staticMembers = new TimelineHashMap<>(snapshotRegistry, 0);
        this.partitionMetadata = new TimelineHashMap<>(snapshotRegistry, 0);
        this.metadataHash = new TimelineLong(snapshotRegistry);
        this.targetAssignmentEpoch = new TimelineInteger(snapshotRegistry);
        this.targetAssignment = new TimelineHashMap<>(snapshotRegistry, 0);
        this.currentActiveTaskToProcessId = new TimelineHashMap<>(snapshotRegistry, 0);
        this.currentStandbyTaskToProcessIds = new TimelineHashMap<>(snapshotRegistry, 0);
        this.currentWarmupTaskToProcessIds = new TimelineHashMap<>(snapshotRegistry, 0);
        this.topology = new TimelineObject<>(snapshotRegistry, Optional.empty());
        this.configuredTopology = new TimelineObject<>(snapshotRegistry, Optional.empty());
    }

    /**
     * @return The group type (Streams).
     */
    @Override
    public GroupType type() {
        return GroupType.STREAMS;