bbejeck commented on code in PR #18729: URL: https://github.com/apache/kafka/pull/18729#discussion_r1945188799
########## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java: ########## @@ -0,0 +1,1180 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.coordinator.group.streams; + +import org.apache.kafka.common.errors.ApiException; +import org.apache.kafka.common.errors.StaleMemberEpochException; +import org.apache.kafka.common.errors.UnknownMemberIdException; +import org.apache.kafka.common.errors.UnsupportedVersionException; +import org.apache.kafka.common.message.ListGroupsResponseData; +import org.apache.kafka.common.message.StreamsGroupDescribeResponseData; +import org.apache.kafka.common.protocol.Errors; +import org.apache.kafka.common.requests.JoinGroupRequest; +import org.apache.kafka.common.utils.LogContext; +import org.apache.kafka.coordinator.common.runtime.CoordinatorRecord; +import org.apache.kafka.coordinator.group.Group; +import org.apache.kafka.coordinator.group.OffsetExpirationCondition; +import org.apache.kafka.coordinator.group.OffsetExpirationConditionImpl; +import org.apache.kafka.coordinator.group.metrics.GroupCoordinatorMetricsShard; +import org.apache.kafka.coordinator.group.streams.topics.ConfiguredTopology; +import org.apache.kafka.coordinator.group.streams.topics.InternalTopicManager; +import org.apache.kafka.image.ClusterImage; +import org.apache.kafka.image.TopicImage; +import org.apache.kafka.image.TopicsImage; +import org.apache.kafka.timeline.SnapshotRegistry; +import org.apache.kafka.timeline.TimelineHashMap; +import org.apache.kafka.timeline.TimelineInteger; +import org.apache.kafka.timeline.TimelineObject; + +import org.slf4j.Logger; + +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; +import java.util.Set; + +import static java.util.Collections.emptyMap; +import static org.apache.kafka.coordinator.group.streams.StreamsGroup.StreamsGroupState.ASSIGNING; +import static org.apache.kafka.coordinator.group.streams.StreamsGroup.StreamsGroupState.EMPTY; +import static org.apache.kafka.coordinator.group.streams.StreamsGroup.StreamsGroupState.NOT_READY; +import static org.apache.kafka.coordinator.group.streams.StreamsGroup.StreamsGroupState.RECONCILING; +import static org.apache.kafka.coordinator.group.streams.StreamsGroup.StreamsGroupState.STABLE; + +/** + * A Streams Group. All the metadata in this class are backed by records in the __consumer_offsets partitions. + */ +public class StreamsGroup implements Group { + + /** + * The protocol type for streams groups. There is only one protocol type, "streams". + */ + private static final String PROTOCOL_TYPE = "streams"; + + public enum StreamsGroupState { + EMPTY("Empty"), + NOT_READY("NotReady"), + ASSIGNING("Assigning"), + RECONCILING("Reconciling"), + STABLE("Stable"), + DEAD("Dead"); + + private final String name; + + private final String lowerCaseName; + + StreamsGroupState(String name) { + this.name = name; + if (Objects.equals(name, "NotReady")) { + this.lowerCaseName = "not_ready"; + } else { + this.lowerCaseName = name.toLowerCase(Locale.ROOT); + } + } + + @Override + public String toString() { + return name; + } + + public String toLowerCaseString() { + return lowerCaseName; + } + } + + public static class DeadlineAndEpoch { + + static final DeadlineAndEpoch EMPTY = new DeadlineAndEpoch(0L, 0); + + public final long deadlineMs; + public final int epoch; + + DeadlineAndEpoch(long deadlineMs, int epoch) { + this.deadlineMs = deadlineMs; + this.epoch = epoch; + } + } + + private final LogContext logContext; + private final Logger log; + + /** + * The snapshot registry. + */ + private final SnapshotRegistry snapshotRegistry; + + /** + * The group id. + */ + private final String groupId; + + /** + * The group state. + */ + private final TimelineObject<StreamsGroupState> state; + + /** + * The group epoch. The epoch is incremented whenever the topology, topic metadata or the set of members changes and it will trigger + * the computation of a new assignment for the group. + */ + private final TimelineInteger groupEpoch; + + /** + * The group members. + */ + private final TimelineHashMap<String, StreamsGroupMember> members; + + /** + * The static group members. + */ + private final TimelineHashMap<String, String> staticMembers; + + /** + * The metadata associated with each subscribed topic name. + */ + private final TimelineHashMap<String, TopicMetadata> partitionMetadata; + + /** + * The target assignment epoch. An assignment epoch smaller than the group epoch means that a new assignment is required. The assignment + * epoch is updated when a new assignment is installed. + */ + private final TimelineInteger targetAssignmentEpoch; + + /** + * The target assignment per member id. + */ + private final TimelineHashMap<String, TasksTuple> targetAssignment; + + /** + * Reverse lookup map representing tasks with their current member assignments. + */ + private final TimelineHashMap<String, TimelineHashMap<Integer, String>> invertedTargetActiveTasksAssignment; + private final TimelineHashMap<String, TimelineHashMap<Integer, String>> invertedTargetStandbyTasksAssignment; + private final TimelineHashMap<String, TimelineHashMap<Integer, String>> invertedTargetWarmupTasksAssignment; + + /** + * These maps map each active/standby/warmup task to the process ID(s) of their current owner. When a + * member revokes a partition, it removes its process ID from this map. When a member gets a partition, it adds its process ID to this map. + */ + private final TimelineHashMap<String, TimelineHashMap<Integer, String>> currentActiveTaskProcessId; + private final TimelineHashMap<String, TimelineHashMap<Integer, Set<String>>> currentStandbyTaskProcessIds; + private final TimelineHashMap<String, TimelineHashMap<Integer, Set<String>>> currentWarmupTaskProcessIds; + + /** + * The coordinator metrics. + */ + private final GroupCoordinatorMetricsShard metrics; + + /** + * The Streams topology. + */ + private final TimelineObject<Optional<StreamsTopology>> topology; + + /** + * 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; + + 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.targetAssignmentEpoch = new TimelineInteger(snapshotRegistry); + this.targetAssignment = new TimelineHashMap<>(snapshotRegistry, 0); + this.invertedTargetActiveTasksAssignment = new TimelineHashMap<>(snapshotRegistry, 0); + this.invertedTargetStandbyTasksAssignment = new TimelineHashMap<>(snapshotRegistry, 0); + this.invertedTargetWarmupTasksAssignment = new TimelineHashMap<>(snapshotRegistry, 0); + this.currentActiveTaskProcessId = new TimelineHashMap<>(snapshotRegistry, 0); + this.currentStandbyTaskProcessIds = new TimelineHashMap<>(snapshotRegistry, 0); + this.currentWarmupTaskProcessIds = new TimelineHashMap<>(snapshotRegistry, 0); + this.metrics = Objects.requireNonNull(metrics); + 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; + } + + /** + * @return The current state as a String. + */ + @Override + public String stateAsString() { + return state.get().toString(); + } + + /** + * @return The current state as a String with given committedOffset. + */ + public String stateAsString(long committedOffset) { + return state.get(committedOffset).toString(); + } + + /** + * @return the group formatted as a list group response based on the committed offset. + */ + public ListGroupsResponseData.ListedGroup asListedGroup(long committedOffset) { + return new ListGroupsResponseData.ListedGroup() + .setGroupId(groupId) + .setProtocolType(PROTOCOL_TYPE) + .setGroupState(state.get(committedOffset).toString()) + .setGroupType(type().toString()); + } + + public ConfiguredTopology configuredTopology() { + return configuredTopology.get().orElse(null); + } + + public StreamsTopology topology() { + return topology.get().orElse(null); + } + + public void setTopology(StreamsTopology topology) { + this.topology.set(Optional.of(topology)); + maybeUpdateConfiguredTopology(); + maybeUpdateGroupState(); + } + + /** + * @return The group id. + */ + @Override + public String groupId() { + return groupId; + } + + /** + * @return The current state. + */ + public StreamsGroupState state() { + return state.get(); + } + + /** + * @return The current state based on committed offset. + */ + public StreamsGroupState state(long committedOffset) { Review Comment: When is this method called? How often is the offset updated? From the parameter name, I'm inferring that it occurs after every commit ########## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java: ########## @@ -0,0 +1,1180 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.coordinator.group.streams; + +import org.apache.kafka.common.errors.ApiException; +import org.apache.kafka.common.errors.StaleMemberEpochException; +import org.apache.kafka.common.errors.UnknownMemberIdException; +import org.apache.kafka.common.errors.UnsupportedVersionException; +import org.apache.kafka.common.message.ListGroupsResponseData; +import org.apache.kafka.common.message.StreamsGroupDescribeResponseData; +import org.apache.kafka.common.protocol.Errors; +import org.apache.kafka.common.requests.JoinGroupRequest; +import org.apache.kafka.common.utils.LogContext; +import org.apache.kafka.coordinator.common.runtime.CoordinatorRecord; +import org.apache.kafka.coordinator.group.Group; +import org.apache.kafka.coordinator.group.OffsetExpirationCondition; +import org.apache.kafka.coordinator.group.OffsetExpirationConditionImpl; +import org.apache.kafka.coordinator.group.metrics.GroupCoordinatorMetricsShard; +import org.apache.kafka.coordinator.group.streams.topics.ConfiguredTopology; +import org.apache.kafka.coordinator.group.streams.topics.InternalTopicManager; +import org.apache.kafka.image.ClusterImage; +import org.apache.kafka.image.TopicImage; +import org.apache.kafka.image.TopicsImage; +import org.apache.kafka.timeline.SnapshotRegistry; +import org.apache.kafka.timeline.TimelineHashMap; +import org.apache.kafka.timeline.TimelineInteger; +import org.apache.kafka.timeline.TimelineObject; + +import org.slf4j.Logger; + +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; +import java.util.Set; + +import static java.util.Collections.emptyMap; +import static org.apache.kafka.coordinator.group.streams.StreamsGroup.StreamsGroupState.ASSIGNING; +import static org.apache.kafka.coordinator.group.streams.StreamsGroup.StreamsGroupState.EMPTY; +import static org.apache.kafka.coordinator.group.streams.StreamsGroup.StreamsGroupState.NOT_READY; +import static org.apache.kafka.coordinator.group.streams.StreamsGroup.StreamsGroupState.RECONCILING; +import static org.apache.kafka.coordinator.group.streams.StreamsGroup.StreamsGroupState.STABLE; + +/** + * A Streams Group. All the metadata in this class are backed by records in the __consumer_offsets partitions. + */ +public class StreamsGroup implements Group { + + /** + * The protocol type for streams groups. There is only one protocol type, "streams". + */ + private static final String PROTOCOL_TYPE = "streams"; + + public enum StreamsGroupState { + EMPTY("Empty"), + NOT_READY("NotReady"), + ASSIGNING("Assigning"), + RECONCILING("Reconciling"), + STABLE("Stable"), + DEAD("Dead"); + + private final String name; + + private final String lowerCaseName; + + StreamsGroupState(String name) { + this.name = name; + if (Objects.equals(name, "NotReady")) { + this.lowerCaseName = "not_ready"; + } else { + this.lowerCaseName = name.toLowerCase(Locale.ROOT); + } + } + + @Override + public String toString() { + return name; + } + + public String toLowerCaseString() { + return lowerCaseName; + } + } + + public static class DeadlineAndEpoch { + + static final DeadlineAndEpoch EMPTY = new DeadlineAndEpoch(0L, 0); + + public final long deadlineMs; + public final int epoch; + + DeadlineAndEpoch(long deadlineMs, int epoch) { + this.deadlineMs = deadlineMs; + this.epoch = epoch; + } + } + + private final LogContext logContext; + private final Logger log; + + /** + * The snapshot registry. + */ + private final SnapshotRegistry snapshotRegistry; + + /** + * The group id. + */ + private final String groupId; + + /** + * The group state. + */ + private final TimelineObject<StreamsGroupState> state; + + /** + * The group epoch. The epoch is incremented whenever the topology, topic metadata or the set of members changes and it will trigger + * the computation of a new assignment for the group. + */ + private final TimelineInteger groupEpoch; + + /** + * The group members. + */ + private final TimelineHashMap<String, StreamsGroupMember> members; + + /** + * The static group members. + */ + private final TimelineHashMap<String, String> staticMembers; + + /** + * The metadata associated with each subscribed topic name. + */ + private final TimelineHashMap<String, TopicMetadata> partitionMetadata; + + /** + * The target assignment epoch. An assignment epoch smaller than the group epoch means that a new assignment is required. The assignment + * epoch is updated when a new assignment is installed. + */ + private final TimelineInteger targetAssignmentEpoch; + + /** + * The target assignment per member id. + */ + private final TimelineHashMap<String, TasksTuple> targetAssignment; + + /** + * Reverse lookup map representing tasks with their current member assignments. + */ + private final TimelineHashMap<String, TimelineHashMap<Integer, String>> invertedTargetActiveTasksAssignment; + private final TimelineHashMap<String, TimelineHashMap<Integer, String>> invertedTargetStandbyTasksAssignment; + private final TimelineHashMap<String, TimelineHashMap<Integer, String>> invertedTargetWarmupTasksAssignment; + + /** + * These maps map each active/standby/warmup task to the process ID(s) of their current owner. When a + * member revokes a partition, it removes its process ID from this map. When a member gets a partition, it adds its process ID to this map. + */ + private final TimelineHashMap<String, TimelineHashMap<Integer, String>> currentActiveTaskProcessId; + private final TimelineHashMap<String, TimelineHashMap<Integer, Set<String>>> currentStandbyTaskProcessIds; + private final TimelineHashMap<String, TimelineHashMap<Integer, Set<String>>> currentWarmupTaskProcessIds; Review Comment: I agree with Bruno about a slight renaming for clarity. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
