bbejeck commented on code in PR #18809:
URL: https://github.com/apache/kafka/pull/18809#discussion_r1953463259
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -760,6 +859,43 @@ ConsumerGroup getOrMaybeCreatePersistedConsumerGroup(
}
}
+ /**
+ * The method should be called on the replay path.
+ * Gets or maybe creates a streams group and updates the groups map if a
new group is created.
+ *
+ * @param groupId The group id.
+ * @param createIfNotExists A boolean indicating whether the group should
be
+ * created if it does not exist.
+ *
+ * @return A StreamsGroup.
+ * @throws GroupIdNotFoundException if the group does not exist and
createIfNotExists is false or
+ * if the group is not a streams group.
+ * @throws IllegalStateException if the group does not have the
expected type.
+ * Package private for testing.
+ */
+ private StreamsGroup getOrMaybeCreatePersistedStreamsGroup(
Review Comment:
More of a general question - what's the determining factor that
`getOrMaybeCreateStreamsGroup` or `getOrMaybeCreatePersistedStreamsGroup` gets
called? Apologies for not catching this on the first pass.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetricsShard.java:
##########
@@ -82,7 +82,7 @@ public TimelineGaugeCounter(TimelineLong timelineLong,
AtomicLong atomicLong) {
/**
* Streams group size gauge counters keyed by the metric name.
*/
- private final Map<StreamsGroupState, TimelineGaugeCounter>
streamsGroupGauges;
+ private volatile Map<StreamsGroupState, Long> streamsGroupGauges;
Review Comment:
Why this change (`TimelineGaugeCounter` -> `Long`)? I imagine it's
lighter-weight alternative
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetricsShard.java:
##########
@@ -410,66 +372,4 @@ public void onShareGroupStateTransition(
}
}
}
-
- /**
- * Called when a streams group's state has changed. Increment/decrement
- * the counter accordingly.
- *
- * @param oldState The previous state. null value means that it's a new
group.
- * @param newState The next state. null value means that the group has
been removed.
- */
- public void onStreamsGroupStateTransition(
- StreamsGroupState oldState,
- StreamsGroupState newState
- ) {
- if (newState != null) {
- switch (newState) {
- case EMPTY:
- incrementNumStreamsGroups(StreamsGroupState.EMPTY);
- break;
- case NOT_READY:
- incrementNumStreamsGroups(StreamsGroupState.NOT_READY);
- break;
- case ASSIGNING:
- incrementNumStreamsGroups(StreamsGroupState.ASSIGNING);
- break;
- case RECONCILING:
- incrementNumStreamsGroups(StreamsGroupState.RECONCILING);
- break;
- case STABLE:
- incrementNumStreamsGroups(StreamsGroupState.STABLE);
- break;
- case DEAD:
- incrementNumStreamsGroups(StreamsGroupState.DEAD);
- break;
- default:
Review Comment:
I think I understand why this was removed - do we keep track of these counts
someplace else?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupBuilder.java:
##########
@@ -0,0 +1,116 @@
+/*
+ * 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.coordinator.common.runtime.CoordinatorRecord;
+import org.apache.kafka.coordinator.group.generated.StreamsGroupTopologyValue;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class StreamsGroupBuilder {
+
+ private final String groupId;
+ private final int groupEpoch;
+ private int targetAssignmentEpoch;
+ private StreamsTopology topology;
+ private final Map<String, StreamsGroupMember> members = new HashMap<>();
+ private final Map<String, TasksTuple> targetAssignments = new HashMap<>();
+ private Map<String, TopicMetadata> partitionMetadata = new HashMap<>();
+
+ public StreamsGroupBuilder(String groupId, int groupEpoch) {
+ this.groupId = groupId;
+ this.groupEpoch = groupEpoch;
+ this.targetAssignmentEpoch = 0;
+ this.topology = null;
+ }
+
+ public StreamsGroupBuilder withMember(StreamsGroupMember member) {
+ this.members.put(member.memberId(), member);
+ return this;
+ }
+
+ public StreamsGroupBuilder withPartitionMetadata(Map<String,
TopicMetadata> partitionMetadata) {
Review Comment:
Should we require non-null on any of the builder method parameters?
--
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
