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

Code before:
     * 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;

    /**
     * The assignment configurations for this streams group.
     * This is used to determine when assignment configuration changes should trigger a rebalance.
     */
    private TimelineHashMap<String, String> assignmentConfigs;

    public StreamsGroup(
        LogContext logContext,
        SnapshotRegistry snapshotRegistry,
        String groupId
    ) {
        this.log = logContext.logger(StreamsGroup.class);
        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.validatedTopologyEpoch = new TimelineInteger(snapshotRegistry);
        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());
        this.assignmentConfigs = new TimelineHashMap<>(snapshotRegistry, 0);
        // Set default assignment configuration
        this.assignmentConfigs.put("num.standby.replicas", "0");

Review comment (suggestion):
This may differ from the current assignment configuration for the group, so I wonder if we shouldn't just `lastUsedAssignmentConfig` instead.

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;

    /**
     * The last used assignment configurations for this streams group.
     * This is used to determine when assignment configuration changes should trigger a rebalance.
     */
    private TimelineHashMap<String, String> lastAssignmentConfigs;

    public StreamsGroup(
        LogContext logContext,
        SnapshotRegistry snapshotRegistry,
        String groupId
    ) {
        this.log = logContext.logger(StreamsGroup.class);
        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.validatedTopologyEpoch = new TimelineInteger(snapshotRegistry);
        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());
        this.lastAssignmentConfigs = new TimelineHashMap<>(snapshotRegistry, 0);
    }