You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "lucasbru (via GitHub)" <gi...@apache.org> on 2025/02/05 17:31:10 UTC
[PR] KAFKA-18730: Add replaying streams group state from offset topic [kafka]
lucasbru opened a new pull request, #18809:
URL: https://github.com/apache/kafka/pull/18809
Adds streams group to the GroupMetadataManager, and implements loading
the records from the offset topic into state. The state also contains
two timers (rebalance timeout and session timeout) that are started
after the group coordinator has been loaded.
This is a stacked PR. Review the last commit only.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18730: Add replaying streams group state from offset topic [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #18809:
URL: https://github.com/apache/kafka/pull/18809#discussion_r1958383366
##########
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:
yes, the team working on consumer groups realized that the group counts can become incorrect easily using this approach. Instead, all groups now use a timer that regularly updates the streams group metrics to be equal to the number of groups (instead of incremening/decrementing when groups are created and deleted).
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18730: Add replaying streams group state from offset topic [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18730: Add replaying streams group state from offset topic [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru merged PR #18809:
URL: https://github.com/apache/kafka/pull/18809
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18730: Add replaying streams group state from offset topic [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #18809:
URL: https://github.com/apache/kafka/pull/18809#discussion_r1952406373
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -3138,6 +3274,45 @@ private void replaceMember(
));
}
+ /**
+ * Fences a member from a streams group and maybe downgrade the streams group to a classic group.
+ *
+ * @param group The group.
+ * @param member The member.
+ * @param response The response of the CoordinatorResult.
+ *
+ * @return The CoordinatorResult to be applied.
+ */
+ private <T> CoordinatorResult<T, CoordinatorRecord> streamsGroupFenceMember(
+ StreamsGroup group,
+ StreamsGroupMember member,
+ T response
+ ) {
+ List<CoordinatorRecord> records = new ArrayList<>();
+ removeStreamsMember(records, group.groupId(), member.memberId());
+
+ // We bump the group epoch.
+ int groupEpoch = group.groupEpoch() + 1;
+ records.add(newStreamsGroupEpochRecord(group.groupId(), groupEpoch));
+
+ cancelTimers(group.groupId(), member.memberId());
+
+ return new CoordinatorResult<>(records, response);
+ }
+
+ /**
+ * Write tombstones for the member. The order matters here.
+ *
+ * @param records The list of records to append the member assignment tombstone records.
+ * @param groupId The group id.
+ * @param memberId The member id.
+ */
+ private void removeStreamsMember(List<CoordinatorRecord> records, String groupId, String memberId) {
Review Comment:
nit: Why not return a list of records?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -3329,6 +3550,27 @@ private void scheduleShareGroupSessionTimeout(
);
}
+ /**
+ * Schedules (or reschedules) the session timeout for the member.
+ *
+ * @param groupId The group id.
+ * @param memberId The member id.
+ * @param sessionTimeoutMs The session timeout.
+ */
+ private void scheduleStreamsGroupSessionTimeout(
Review Comment:
This is only used in `scheduleStreamsGroupSessionTimeout(groupId, memberId)`. Will this be used somewhere else in future? Why not calling `timer.schedule()` directly in `scheduleStreamsGroupSessionTimeout(groupId, memberId)`?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorShardTest.java:
##########
@@ -830,6 +844,385 @@ public void testReplayConsumerGroupCurrentMemberAssignmentWithNullValue() {
verify(groupMetadataManager, times(1)).replay(key, null);
}
+ @Test
+ public void testReplayStreamsGroupMetadata() {
+ GroupMetadataManager groupMetadataManager = mock(GroupMetadataManager.class);
+ OffsetMetadataManager offsetMetadataManager = mock(OffsetMetadataManager.class);
+ CoordinatorMetrics coordinatorMetrics = mock(CoordinatorMetrics.class);
+ CoordinatorMetricsShard metricsShard = mock(CoordinatorMetricsShard.class);
+ GroupCoordinatorShard coordinator = new GroupCoordinatorShard(
+ new LogContext(),
+ groupMetadataManager,
+ offsetMetadataManager,
+ Time.SYSTEM,
+ new MockCoordinatorTimer<>(Time.SYSTEM),
+ mock(GroupCoordinatorConfig.class),
+ coordinatorMetrics,
+ metricsShard
+ );
+
+ StreamsGroupMetadataKey key = new StreamsGroupMetadataKey();
+ StreamsGroupMetadataValue value = new StreamsGroupMetadataValue();
+
+ coordinator.replay(0L, RecordBatch.NO_PRODUCER_ID, RecordBatch.NO_PRODUCER_EPOCH, CoordinatorRecord.record(
+ key,
+ new ApiMessageAndVersion(value, (short) 0)
+ ));
+
+ verify(groupMetadataManager, times(1)).replay(key, value);
Review Comment:
You could drop `times(1)` since it is the default. I am fine either way.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -3385,6 +3627,49 @@ private void scheduleConsumerGroupRebalanceTimeout(
});
}
+ /**
+ * Schedules a rebalance timeout for the member.
+ *
+ * @param groupId The group id.
+ * @param memberId The member id.
+ * @param memberEpoch The member epoch.
+ * @param rebalanceTimeoutMs The rebalance timeout.
+ */
+ private void scheduleStreamsGroupRebalanceTimeout(
+ String groupId,
+ String memberId,
+ int memberEpoch,
+ int rebalanceTimeoutMs
+ ) {
+ String key = streamsGroupRebalanceTimeoutKey(groupId, memberId);
+ timer.schedule(key, rebalanceTimeoutMs, TimeUnit.MILLISECONDS, true, () -> {
+ try {
+ StreamsGroup group = streamsGroup(groupId);
+ StreamsGroupMember member = group.getOrMaybeCreateMember(memberId, false);
+
+ if (member.memberEpoch() == memberEpoch) {
+ log.info("[GroupId {}] Member {} fenced from the group because " +
+ "it failed to transition from epoch {} within {}ms.",
+ groupId, memberId, memberEpoch, rebalanceTimeoutMs);
+
+ return streamsGroupFenceMember(group, member, null);
+ } else {
+ log.debug("[GroupId {}] Ignoring rebalance timeout for {} because the member " +
+ "left the epoch {}.", groupId, memberId, memberEpoch);
Review Comment:
What does "leave the epoch" mean?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -15135,6 +15284,229 @@ public void testReplayConsumerGroupCurrentMemberAssignmentTombstone() {
assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.consumerGroup("bar"));
}
+ @Test
+ public void testReplayStreamsGroupMemberMetadata() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ StreamsGroupMember member = new StreamsGroupMember.Builder("member")
+ .setClientId("clientid")
+ .setClientHost("clienthost")
+ .setRackId("rackid")
+ .setInstanceId("instanceid")
+ .setRebalanceTimeoutMs(1000)
+ .setTopologyEpoch(10)
+ .setProcessId("processid")
+ .setUserEndpoint(new Endpoint().setHost("localhost").setPort(9999))
+ .setClientTags(Collections.singletonMap("key", "value"))
+ .build();
+
+ // The group and the member are created if they do not exist.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMemberRecord("foo", member));
+ assertEquals(member, context.groupMetadataManager.streamsGroup("foo").getOrMaybeCreateMember("member", false));
+ }
+
+ @Test
+ public void testReplayStreamsGroupMemberMetadataTombstone() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group still exists but the member is already gone. Replaying the
+ // StreamsGroupMemberMetadata tombstone should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupEpochRecord("foo", 10));
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMemberTombstoneRecord("foo", "m1"));
+ assertThrows(UnknownMemberIdException.class, () -> context.groupMetadataManager.streamsGroup("foo").getOrMaybeCreateMember("m1", false));
+
+ // The group may not exist at all. Replaying the StreamsGroupMemberMetadata tombstone
+ // should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMemberTombstoneRecord("bar", "m1"));
+ assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.streamsGroup("bar"));
+ }
+
+ @Test
+ public void testReplayStreamsGroupMetadata() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group is created if it does not exist.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupEpochRecord("foo", 10));
+ assertEquals(10, context.groupMetadataManager.streamsGroup("foo").groupEpoch());
+ }
+
+ @Test
+ public void testReplayStreamsGroupMetadataTombstone() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group may not exist at all. Replaying the StreamsGroupMetadata tombstone
+ // should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupEpochTombstoneRecord("foo"));
+ assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.streamsGroup("foo"));
+ }
+
+ @Test
+ public void testReplayStreamsGroupPartitionMetadata() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ Map<String, org.apache.kafka.coordinator.group.streams.TopicMetadata> metadata = Map.of(
+ "bar",
+ new org.apache.kafka.coordinator.group.streams.TopicMetadata(Uuid.randomUuid(), "bar", 10)
+ );
+
+ // The group is created if it does not exist.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupPartitionMetadataRecord("foo", metadata));
+ assertEquals(metadata, context.groupMetadataManager.streamsGroup("foo").partitionMetadata());
+ }
+
+ @Test
+ public void testReplayStreamsGroupPartitionMetadataTombstone() {
Review Comment:
Could you please also add a test that removes an existing group?
What about all the `IllegalStateException` that are thrown in the method under test?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -3138,6 +3274,45 @@ private void replaceMember(
));
}
+ /**
+ * Fences a member from a streams group and maybe downgrade the streams group to a classic group.
Review Comment:
Is the classic part true for Streams?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -15135,6 +15284,229 @@ public void testReplayConsumerGroupCurrentMemberAssignmentTombstone() {
assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.consumerGroup("bar"));
}
+ @Test
+ public void testReplayStreamsGroupMemberMetadata() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ StreamsGroupMember member = new StreamsGroupMember.Builder("member")
+ .setClientId("clientid")
+ .setClientHost("clienthost")
+ .setRackId("rackid")
+ .setInstanceId("instanceid")
+ .setRebalanceTimeoutMs(1000)
+ .setTopologyEpoch(10)
+ .setProcessId("processid")
+ .setUserEndpoint(new Endpoint().setHost("localhost").setPort(9999))
+ .setClientTags(Collections.singletonMap("key", "value"))
+ .build();
+
+ // The group and the member are created if they do not exist.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMemberRecord("foo", member));
+ assertEquals(member, context.groupMetadataManager.streamsGroup("foo").getOrMaybeCreateMember("member", false));
+ }
+
+ @Test
+ public void testReplayStreamsGroupMemberMetadataTombstone() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group still exists but the member is already gone. Replaying the
+ // StreamsGroupMemberMetadata tombstone should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupEpochRecord("foo", 10));
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMemberTombstoneRecord("foo", "m1"));
+ assertThrows(UnknownMemberIdException.class, () -> context.groupMetadataManager.streamsGroup("foo").getOrMaybeCreateMember("m1", false));
+
+ // The group may not exist at all. Replaying the StreamsGroupMemberMetadata tombstone
+ // should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMemberTombstoneRecord("bar", "m1"));
+ assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.streamsGroup("bar"));
+ }
+
+ @Test
+ public void testReplayStreamsGroupMetadata() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group is created if it does not exist.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupEpochRecord("foo", 10));
+ assertEquals(10, context.groupMetadataManager.streamsGroup("foo").groupEpoch());
+ }
+
+ @Test
+ public void testReplayStreamsGroupMetadataTombstone() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group may not exist at all. Replaying the StreamsGroupMetadata tombstone
+ // should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupEpochTombstoneRecord("foo"));
+ assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.streamsGroup("foo"));
+ }
+
+ @Test
+ public void testReplayStreamsGroupPartitionMetadata() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ Map<String, org.apache.kafka.coordinator.group.streams.TopicMetadata> metadata = Map.of(
+ "bar",
+ new org.apache.kafka.coordinator.group.streams.TopicMetadata(Uuid.randomUuid(), "bar", 10)
+ );
+
+ // The group is created if it does not exist.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupPartitionMetadataRecord("foo", metadata));
+ assertEquals(metadata, context.groupMetadataManager.streamsGroup("foo").partitionMetadata());
+ }
+
+ @Test
+ public void testReplayStreamsGroupPartitionMetadataTombstone() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group may not exist at all. Replaying the StreamsGroupPartitionMetadata tombstone
+ // should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupPartitionMetadataTombstoneRecord("foo"));
+ assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.streamsGroup("foo"));
+ }
+
+ @Test
+ public void testReplayStreamsGroupTargetAssignmentMember() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group is created if it does not exist.
+ final TasksTuple tasks =
+ new TasksTuple(
+ TaskAssignmentTestUtil.mkTasksPerSubtopology(
+ TaskAssignmentTestUtil.mkTasks("subtopology-1", 0, 1, 2)),
+ TaskAssignmentTestUtil.mkTasksPerSubtopology(
+ TaskAssignmentTestUtil.mkTasks("subtopology-1", 3, 4, 5)),
+ TaskAssignmentTestUtil.mkTasksPerSubtopology(
+ TaskAssignmentTestUtil.mkTasks("subtopology-1", 6, 7, 8))
+ );
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentRecord("foo", "m1", tasks));
+ assertEquals(tasks.activeTasks(), context.groupMetadataManager.streamsGroup("foo").targetAssignment("m1").activeTasks());
+ assertEquals(tasks.standbyTasks(), context.groupMetadataManager.streamsGroup("foo").targetAssignment("m1").standbyTasks());
+ assertEquals(tasks.warmupTasks(), context.groupMetadataManager.streamsGroup("foo").targetAssignment("m1").warmupTasks());
+ }
+
+ @Test
+ public void testReplayStreamsGroupTargetAssignmentMemberTombstone() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group may not exist at all. Replaying the StreamsGroupTargetAssignmentMember tombstone
+ // should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentTombstoneRecord("foo", "m1"));
+ assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.streamsGroup("foo"));
+ }
+
+ @Test
+ public void testReplayStreamsGroupTargetAssignmentMetadata() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group is created if it does not exist.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentEpochRecord("foo", 10));
+ assertEquals(10, context.groupMetadataManager.streamsGroup("foo").assignmentEpoch());
+ }
+
+ @Test
+ public void testReplayStreamsGroupTargetAssignmentMetadataTombstone() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group may not exist at all. Replaying the StreamsGroupTargetAssignmentMetadata tombstone
+ // should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentEpochTombstoneRecord("foo"));
+ assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.streamsGroup("foo"));
+ }
+
+ @Test
+ public void testReplayStreamsGroupCurrentMemberAssignment() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ StreamsGroupMember member = new StreamsGroupMember.Builder("member")
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setState(org.apache.kafka.coordinator.group.streams.MemberState.UNRELEASED_TASKS)
+ .setAssignedTasks(new TasksTuple(
+ TaskAssignmentTestUtil.mkTasksPerSubtopology(TaskAssignmentTestUtil.mkTasks("subtopology-1", 0, 1, 2)),
+ TaskAssignmentTestUtil.mkTasksPerSubtopology(TaskAssignmentTestUtil.mkTasks("subtopology-1", 3, 4, 5)),
+ TaskAssignmentTestUtil.mkTasksPerSubtopology(TaskAssignmentTestUtil.mkTasks("subtopology-1", 6, 7, 8))
+ ))
+ .setTasksPendingRevocation(TasksTuple.EMPTY)
+ .build();
+
+ // The group and the member are created if they do not exist.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupCurrentAssignmentRecord("bar", member));
+ assertEquals(member, context.groupMetadataManager.streamsGroup("bar").getOrMaybeCreateMember("member", false));
+ }
+
+ @Test
+ public void testReplayStreamsGroupCurrentMemberAssignmentTombstone() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group still exists, but the member is already gone. Replaying the
+ // StreamsGroupCurrentMemberAssignment tombstone should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupEpochRecord("foo", 10));
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupCurrentAssignmentTombstoneRecord("foo", "m1"));
+ assertThrows(UnknownMemberIdException.class, () -> context.groupMetadataManager.streamsGroup("foo").getOrMaybeCreateMember("m1", false));
+
+ // The group may not exist at all. Replaying the StreamsGroupCurrentMemberAssignment tombstone
+ // should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupCurrentAssignmentTombstoneRecord("bar", "m1"));
+ assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.streamsGroup("bar"));
+ }
+
+ @Test
+ public void testReplayStreamsGroupTopology() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ StreamsGroupTopologyValue topology = new StreamsGroupTopologyValue()
+ .setEpoch(12)
+ .setSubtopologies(
+ List.of(
+ new StreamsGroupTopologyValue.Subtopology()
+ .setSubtopologyId("subtopology-1")
+ .setSourceTopics(List.of("source-topic"))
+ .setRepartitionSinkTopics(List.of("sink-topic"))
+ )
+ );
+
+ // The group and the topology are created if they do not exist.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupTopologyRecord("bar", topology));
+ final Optional<StreamsTopology> actualTopology = context.groupMetadataManager.streamsGroup("bar").topology();
+ assertTrue(actualTopology.isPresent(), "topology should be set");
+ assertEquals(topology.epoch(), actualTopology.get().topologyEpoch());
+ assertEquals(topology.subtopologies().size(), actualTopology.get().subtopologies().size());
+ assertEquals(
+ topology.subtopologies().iterator().next(),
+ actualTopology.get().subtopologies().values().iterator().next()
+ );
+ }
+
+ @Test
+ public void testReplayStreamsGroupTopologyTombstone() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
Review Comment:
Could you please add a test for when a group exists?
##########
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.
+ */
+ StreamsGroup getOrMaybeCreatePersistedStreamsGroup(
Review Comment:
why is this method package-private? It is not used outside of this class.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -15135,6 +15284,229 @@ public void testReplayConsumerGroupCurrentMemberAssignmentTombstone() {
assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.consumerGroup("bar"));
}
+ @Test
+ public void testReplayStreamsGroupMemberMetadata() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ StreamsGroupMember member = new StreamsGroupMember.Builder("member")
+ .setClientId("clientid")
+ .setClientHost("clienthost")
+ .setRackId("rackid")
+ .setInstanceId("instanceid")
+ .setRebalanceTimeoutMs(1000)
+ .setTopologyEpoch(10)
+ .setProcessId("processid")
+ .setUserEndpoint(new Endpoint().setHost("localhost").setPort(9999))
+ .setClientTags(Collections.singletonMap("key", "value"))
+ .build();
+
+ // The group and the member are created if they do not exist.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMemberRecord("foo", member));
+ assertEquals(member, context.groupMetadataManager.streamsGroup("foo").getOrMaybeCreateMember("member", false));
+ }
+
+ @Test
+ public void testReplayStreamsGroupMemberMetadataTombstone() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group still exists but the member is already gone. Replaying the
+ // StreamsGroupMemberMetadata tombstone should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupEpochRecord("foo", 10));
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMemberTombstoneRecord("foo", "m1"));
+ assertThrows(UnknownMemberIdException.class, () -> context.groupMetadataManager.streamsGroup("foo").getOrMaybeCreateMember("m1", false));
+
+ // The group may not exist at all. Replaying the StreamsGroupMemberMetadata tombstone
+ // should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMemberTombstoneRecord("bar", "m1"));
+ assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.streamsGroup("bar"));
+ }
+
+ @Test
+ public void testReplayStreamsGroupMetadata() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group is created if it does not exist.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupEpochRecord("foo", 10));
+ assertEquals(10, context.groupMetadataManager.streamsGroup("foo").groupEpoch());
+ }
+
+ @Test
+ public void testReplayStreamsGroupMetadataTombstone() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group may not exist at all. Replaying the StreamsGroupMetadata tombstone
+ // should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupEpochTombstoneRecord("foo"));
+ assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.streamsGroup("foo"));
+ }
+
+ @Test
+ public void testReplayStreamsGroupPartitionMetadata() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ Map<String, org.apache.kafka.coordinator.group.streams.TopicMetadata> metadata = Map.of(
+ "bar",
+ new org.apache.kafka.coordinator.group.streams.TopicMetadata(Uuid.randomUuid(), "bar", 10)
+ );
+
+ // The group is created if it does not exist.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupPartitionMetadataRecord("foo", metadata));
+ assertEquals(metadata, context.groupMetadataManager.streamsGroup("foo").partitionMetadata());
+ }
+
+ @Test
+ public void testReplayStreamsGroupPartitionMetadataTombstone() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group may not exist at all. Replaying the StreamsGroupPartitionMetadata tombstone
+ // should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupPartitionMetadataTombstoneRecord("foo"));
+ assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.streamsGroup("foo"));
+ }
+
+ @Test
+ public void testReplayStreamsGroupTargetAssignmentMember() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group is created if it does not exist.
+ final TasksTuple tasks =
+ new TasksTuple(
+ TaskAssignmentTestUtil.mkTasksPerSubtopology(
+ TaskAssignmentTestUtil.mkTasks("subtopology-1", 0, 1, 2)),
+ TaskAssignmentTestUtil.mkTasksPerSubtopology(
+ TaskAssignmentTestUtil.mkTasks("subtopology-1", 3, 4, 5)),
+ TaskAssignmentTestUtil.mkTasksPerSubtopology(
+ TaskAssignmentTestUtil.mkTasks("subtopology-1", 6, 7, 8))
+ );
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentRecord("foo", "m1", tasks));
+ assertEquals(tasks.activeTasks(), context.groupMetadataManager.streamsGroup("foo").targetAssignment("m1").activeTasks());
+ assertEquals(tasks.standbyTasks(), context.groupMetadataManager.streamsGroup("foo").targetAssignment("m1").standbyTasks());
+ assertEquals(tasks.warmupTasks(), context.groupMetadataManager.streamsGroup("foo").targetAssignment("m1").warmupTasks());
+ }
+
+ @Test
+ public void testReplayStreamsGroupTargetAssignmentMemberTombstone() {
Review Comment:
Could you please add a test for when a group exists?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorShardTest.java:
##########
@@ -830,6 +844,385 @@ public void testReplayConsumerGroupCurrentMemberAssignmentWithNullValue() {
verify(groupMetadataManager, times(1)).replay(key, null);
}
+ @Test
+ public void testReplayStreamsGroupMetadata() {
+ GroupMetadataManager groupMetadataManager = mock(GroupMetadataManager.class);
+ OffsetMetadataManager offsetMetadataManager = mock(OffsetMetadataManager.class);
+ CoordinatorMetrics coordinatorMetrics = mock(CoordinatorMetrics.class);
+ CoordinatorMetricsShard metricsShard = mock(CoordinatorMetricsShard.class);
+ GroupCoordinatorShard coordinator = new GroupCoordinatorShard(
+ new LogContext(),
+ groupMetadataManager,
+ offsetMetadataManager,
+ Time.SYSTEM,
+ new MockCoordinatorTimer<>(Time.SYSTEM),
+ mock(GroupCoordinatorConfig.class),
+ coordinatorMetrics,
+ metricsShard
+ );
+
Review Comment:
nit: I would delete this line so that the call under test is more easily identifiable (also in the other tests). Just a proposal. Feel free to ignore if you want.
```suggestion
```
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -3962,7 +4359,167 @@ public void replay(
}
removeGroup(groupId);
}
+ }
+
+ /**
+ * Replays StreamsGroupMemberMetadataKey/Value to update the hard state of
+ * the streams group.
+ * It updates the subscription part of the member or deletes the member.
+ *
+ * @param key A StreamsGroupMemberMetadataKey key.
+ * @param value A StreamsGroupMemberMetadataValue record.
+ */
+ public void replay(
+ StreamsGroupMemberMetadataKey key,
+ StreamsGroupMemberMetadataValue value
+ ) {
+ String groupId = key.groupId();
+ String memberId = key.memberId();
+
+ StreamsGroup streamsGroup;
+ try {
+ streamsGroup = getOrMaybeCreatePersistedStreamsGroup(groupId, value != null);
+ } catch (GroupIdNotFoundException ex) {
+ // If the group does not exist and a tombstone is replayed, we can ignore it.
+ return;
+ }
+
+ if (value != null) {
+ StreamsGroupMember oldMember = streamsGroup.getOrMaybeCreateMember(memberId, true);
+ streamsGroup.updateMember(new StreamsGroupMember.Builder(oldMember)
+ .updateWith(value)
+ .build());
Review Comment:
I find it confusing that the Streams group object creates a member without adding it to its members and the member needs to be added with `updateMember()`. I know that it is done this way for the other groups. Creating a member outside of the group should not be a concern of the group.
You do not need to change anything. Just a thought that I had for discussion.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -15135,6 +15284,229 @@ public void testReplayConsumerGroupCurrentMemberAssignmentTombstone() {
assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.consumerGroup("bar"));
}
+ @Test
+ public void testReplayStreamsGroupMemberMetadata() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ StreamsGroupMember member = new StreamsGroupMember.Builder("member")
+ .setClientId("clientid")
+ .setClientHost("clienthost")
+ .setRackId("rackid")
+ .setInstanceId("instanceid")
+ .setRebalanceTimeoutMs(1000)
+ .setTopologyEpoch(10)
+ .setProcessId("processid")
+ .setUserEndpoint(new Endpoint().setHost("localhost").setPort(9999))
+ .setClientTags(Collections.singletonMap("key", "value"))
+ .build();
+
+ // The group and the member are created if they do not exist.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMemberRecord("foo", member));
+ assertEquals(member, context.groupMetadataManager.streamsGroup("foo").getOrMaybeCreateMember("member", false));
+ }
+
+ @Test
+ public void testReplayStreamsGroupMemberMetadataTombstone() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group still exists but the member is already gone. Replaying the
+ // StreamsGroupMemberMetadata tombstone should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupEpochRecord("foo", 10));
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMemberTombstoneRecord("foo", "m1"));
+ assertThrows(UnknownMemberIdException.class, () -> context.groupMetadataManager.streamsGroup("foo").getOrMaybeCreateMember("m1", false));
+
+ // The group may not exist at all. Replaying the StreamsGroupMemberMetadata tombstone
+ // should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMemberTombstoneRecord("bar", "m1"));
+ assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.streamsGroup("bar"));
+ }
+
+ @Test
+ public void testReplayStreamsGroupMetadata() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group is created if it does not exist.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupEpochRecord("foo", 10));
+ assertEquals(10, context.groupMetadataManager.streamsGroup("foo").groupEpoch());
+ }
+
+ @Test
+ public void testReplayStreamsGroupMetadataTombstone() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group may not exist at all. Replaying the StreamsGroupMetadata tombstone
+ // should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupEpochTombstoneRecord("foo"));
+ assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.streamsGroup("foo"));
+ }
+
+ @Test
+ public void testReplayStreamsGroupPartitionMetadata() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ Map<String, org.apache.kafka.coordinator.group.streams.TopicMetadata> metadata = Map.of(
+ "bar",
+ new org.apache.kafka.coordinator.group.streams.TopicMetadata(Uuid.randomUuid(), "bar", 10)
+ );
+
+ // The group is created if it does not exist.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupPartitionMetadataRecord("foo", metadata));
+ assertEquals(metadata, context.groupMetadataManager.streamsGroup("foo").partitionMetadata());
+ }
+
+ @Test
+ public void testReplayStreamsGroupPartitionMetadataTombstone() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group may not exist at all. Replaying the StreamsGroupPartitionMetadata tombstone
+ // should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupPartitionMetadataTombstoneRecord("foo"));
+ assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.streamsGroup("foo"));
+ }
+
+ @Test
+ public void testReplayStreamsGroupTargetAssignmentMember() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group is created if it does not exist.
+ final TasksTuple tasks =
+ new TasksTuple(
+ TaskAssignmentTestUtil.mkTasksPerSubtopology(
+ TaskAssignmentTestUtil.mkTasks("subtopology-1", 0, 1, 2)),
+ TaskAssignmentTestUtil.mkTasksPerSubtopology(
+ TaskAssignmentTestUtil.mkTasks("subtopology-1", 3, 4, 5)),
+ TaskAssignmentTestUtil.mkTasksPerSubtopology(
+ TaskAssignmentTestUtil.mkTasks("subtopology-1", 6, 7, 8))
+ );
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentRecord("foo", "m1", tasks));
+ assertEquals(tasks.activeTasks(), context.groupMetadataManager.streamsGroup("foo").targetAssignment("m1").activeTasks());
+ assertEquals(tasks.standbyTasks(), context.groupMetadataManager.streamsGroup("foo").targetAssignment("m1").standbyTasks());
+ assertEquals(tasks.warmupTasks(), context.groupMetadataManager.streamsGroup("foo").targetAssignment("m1").warmupTasks());
+ }
+
+ @Test
+ public void testReplayStreamsGroupTargetAssignmentMemberTombstone() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group may not exist at all. Replaying the StreamsGroupTargetAssignmentMember tombstone
+ // should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentTombstoneRecord("foo", "m1"));
+ assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.streamsGroup("foo"));
+ }
+
+ @Test
+ public void testReplayStreamsGroupTargetAssignmentMetadata() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group is created if it does not exist.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentEpochRecord("foo", 10));
+ assertEquals(10, context.groupMetadataManager.streamsGroup("foo").assignmentEpoch());
+ }
+
+ @Test
+ public void testReplayStreamsGroupTargetAssignmentMetadataTombstone() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group may not exist at all. Replaying the StreamsGroupTargetAssignmentMetadata tombstone
+ // should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentEpochTombstoneRecord("foo"));
+ assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.streamsGroup("foo"));
+ }
+
+ @Test
+ public void testReplayStreamsGroupCurrentMemberAssignment() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ StreamsGroupMember member = new StreamsGroupMember.Builder("member")
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setState(org.apache.kafka.coordinator.group.streams.MemberState.UNRELEASED_TASKS)
+ .setAssignedTasks(new TasksTuple(
+ TaskAssignmentTestUtil.mkTasksPerSubtopology(TaskAssignmentTestUtil.mkTasks("subtopology-1", 0, 1, 2)),
+ TaskAssignmentTestUtil.mkTasksPerSubtopology(TaskAssignmentTestUtil.mkTasks("subtopology-1", 3, 4, 5)),
+ TaskAssignmentTestUtil.mkTasksPerSubtopology(TaskAssignmentTestUtil.mkTasks("subtopology-1", 6, 7, 8))
+ ))
+ .setTasksPendingRevocation(TasksTuple.EMPTY)
+ .build();
+
+ // The group and the member are created if they do not exist.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupCurrentAssignmentRecord("bar", member));
+ assertEquals(member, context.groupMetadataManager.streamsGroup("bar").getOrMaybeCreateMember("member", false));
+ }
+
+ @Test
+ public void testReplayStreamsGroupCurrentMemberAssignmentTombstone() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
Review Comment:
Could you please add a test for when a group exists?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -687,6 +717,75 @@ ConsumerGroup getOrMaybeCreateConsumerGroup(
}
}
+ /**
+ * Gets or maybe creates a streams group without updating the groups map.
+ * The group will be materialized during the replay.
+ *
+ * @param groupId The group id.
+ * @param createIfNotExists A boolean indicating whether the group should be
+ * created if it does not exist or is an empty classic group.
+ *
+ * @return A StreamsGroup.
+ * @throws GroupIdNotFoundException if the group does not exist and createIfNotExists is false or
+ * if the group is not a streams group.
+ *
+ * Package private for testing.
+ */
+ StreamsGroup getOrMaybeCreateStreamsGroup(
+ String groupId,
+ boolean createIfNotExists
+ ) throws GroupIdNotFoundException {
+ Group group = groups.get(groupId);
+
+ if (group == null && !createIfNotExists) {
+ throw new GroupIdNotFoundException(String.format("Streams group %s not found.", groupId));
+ }
+
+ if (group == null) {
+ return new StreamsGroup(logContext, snapshotRegistry, groupId, metrics);
+ } else {
+ if (group.type() == STREAMS) {
+ return (StreamsGroup) group;
+ } else {
+ throw new GroupIdNotFoundException(String.format("Group %s is not a streams group.", groupId));
+ }
+ }
+ }
+
+ /**
+ * Gets a streams group by committed offset.
+ *
+ * @param groupId The group id.
+ * @param committedOffset A specified committed offset corresponding to this shard.
+ *
+ * @return A StreamsGroup.
+ * @throws GroupIdNotFoundException if the group does not exist or is not a streams group.
+ */
+ public StreamsGroup streamsGroup(
Review Comment:
Why is this method `public`? It is not used outside of this class.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -15135,6 +15284,229 @@ public void testReplayConsumerGroupCurrentMemberAssignmentTombstone() {
assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.consumerGroup("bar"));
}
+ @Test
+ public void testReplayStreamsGroupMemberMetadata() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ StreamsGroupMember member = new StreamsGroupMember.Builder("member")
+ .setClientId("clientid")
+ .setClientHost("clienthost")
+ .setRackId("rackid")
+ .setInstanceId("instanceid")
+ .setRebalanceTimeoutMs(1000)
+ .setTopologyEpoch(10)
+ .setProcessId("processid")
+ .setUserEndpoint(new Endpoint().setHost("localhost").setPort(9999))
+ .setClientTags(Collections.singletonMap("key", "value"))
+ .build();
+
+ // The group and the member are created if they do not exist.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMemberRecord("foo", member));
+ assertEquals(member, context.groupMetadataManager.streamsGroup("foo").getOrMaybeCreateMember("member", false));
+ }
+
+ @Test
+ public void testReplayStreamsGroupMemberMetadataTombstone() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group still exists but the member is already gone. Replaying the
+ // StreamsGroupMemberMetadata tombstone should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupEpochRecord("foo", 10));
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMemberTombstoneRecord("foo", "m1"));
+ assertThrows(UnknownMemberIdException.class, () -> context.groupMetadataManager.streamsGroup("foo").getOrMaybeCreateMember("m1", false));
+
+ // The group may not exist at all. Replaying the StreamsGroupMemberMetadata tombstone
+ // should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMemberTombstoneRecord("bar", "m1"));
+ assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.streamsGroup("bar"));
+ }
+
+ @Test
+ public void testReplayStreamsGroupMetadata() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group is created if it does not exist.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupEpochRecord("foo", 10));
+ assertEquals(10, context.groupMetadataManager.streamsGroup("foo").groupEpoch());
+ }
+
+ @Test
+ public void testReplayStreamsGroupMetadataTombstone() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group may not exist at all. Replaying the StreamsGroupMetadata tombstone
+ // should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupEpochTombstoneRecord("foo"));
+ assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.streamsGroup("foo"));
+ }
+
+ @Test
+ public void testReplayStreamsGroupPartitionMetadata() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ Map<String, org.apache.kafka.coordinator.group.streams.TopicMetadata> metadata = Map.of(
+ "bar",
+ new org.apache.kafka.coordinator.group.streams.TopicMetadata(Uuid.randomUuid(), "bar", 10)
+ );
+
+ // The group is created if it does not exist.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupPartitionMetadataRecord("foo", metadata));
+ assertEquals(metadata, context.groupMetadataManager.streamsGroup("foo").partitionMetadata());
+ }
+
+ @Test
+ public void testReplayStreamsGroupPartitionMetadataTombstone() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group may not exist at all. Replaying the StreamsGroupPartitionMetadata tombstone
+ // should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupPartitionMetadataTombstoneRecord("foo"));
+ assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.streamsGroup("foo"));
+ }
+
+ @Test
+ public void testReplayStreamsGroupTargetAssignmentMember() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group is created if it does not exist.
+ final TasksTuple tasks =
+ new TasksTuple(
+ TaskAssignmentTestUtil.mkTasksPerSubtopology(
+ TaskAssignmentTestUtil.mkTasks("subtopology-1", 0, 1, 2)),
+ TaskAssignmentTestUtil.mkTasksPerSubtopology(
+ TaskAssignmentTestUtil.mkTasks("subtopology-1", 3, 4, 5)),
+ TaskAssignmentTestUtil.mkTasksPerSubtopology(
+ TaskAssignmentTestUtil.mkTasks("subtopology-1", 6, 7, 8))
+ );
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentRecord("foo", "m1", tasks));
+ assertEquals(tasks.activeTasks(), context.groupMetadataManager.streamsGroup("foo").targetAssignment("m1").activeTasks());
+ assertEquals(tasks.standbyTasks(), context.groupMetadataManager.streamsGroup("foo").targetAssignment("m1").standbyTasks());
+ assertEquals(tasks.warmupTasks(), context.groupMetadataManager.streamsGroup("foo").targetAssignment("m1").warmupTasks());
+ }
+
+ @Test
+ public void testReplayStreamsGroupTargetAssignmentMemberTombstone() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group may not exist at all. Replaying the StreamsGroupTargetAssignmentMember tombstone
+ // should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentTombstoneRecord("foo", "m1"));
+ assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.streamsGroup("foo"));
+ }
+
+ @Test
+ public void testReplayStreamsGroupTargetAssignmentMetadata() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group is created if it does not exist.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentEpochRecord("foo", 10));
+ assertEquals(10, context.groupMetadataManager.streamsGroup("foo").assignmentEpoch());
+ }
+
+ @Test
+ public void testReplayStreamsGroupTargetAssignmentMetadataTombstone() {
Review Comment:
Could you please add a test for when a group exists?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18730: Add replaying streams group state from offset topic [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #18809:
URL: https://github.com/apache/kafka/pull/18809#discussion_r1952699451
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -3138,6 +3274,45 @@ private void replaceMember(
));
}
+ /**
+ * Fences a member from a streams group and maybe downgrade the streams group to a classic group.
Review Comment:
Done
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorShardTest.java:
##########
@@ -830,6 +844,385 @@ public void testReplayConsumerGroupCurrentMemberAssignmentWithNullValue() {
verify(groupMetadataManager, times(1)).replay(key, null);
}
+ @Test
+ public void testReplayStreamsGroupMetadata() {
+ GroupMetadataManager groupMetadataManager = mock(GroupMetadataManager.class);
+ OffsetMetadataManager offsetMetadataManager = mock(OffsetMetadataManager.class);
+ CoordinatorMetrics coordinatorMetrics = mock(CoordinatorMetrics.class);
+ CoordinatorMetricsShard metricsShard = mock(CoordinatorMetricsShard.class);
+ GroupCoordinatorShard coordinator = new GroupCoordinatorShard(
+ new LogContext(),
+ groupMetadataManager,
+ offsetMetadataManager,
+ Time.SYSTEM,
+ new MockCoordinatorTimer<>(Time.SYSTEM),
+ mock(GroupCoordinatorConfig.class),
+ coordinatorMetrics,
+ metricsShard
+ );
+
Review Comment:
Done
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -3385,6 +3627,49 @@ private void scheduleConsumerGroupRebalanceTimeout(
});
}
+ /**
+ * Schedules a rebalance timeout for the member.
+ *
+ * @param groupId The group id.
+ * @param memberId The member id.
+ * @param memberEpoch The member epoch.
+ * @param rebalanceTimeoutMs The rebalance timeout.
+ */
+ private void scheduleStreamsGroupRebalanceTimeout(
+ String groupId,
+ String memberId,
+ int memberEpoch,
+ int rebalanceTimeoutMs
+ ) {
+ String key = streamsGroupRebalanceTimeoutKey(groupId, memberId);
+ timer.schedule(key, rebalanceTimeoutMs, TimeUnit.MILLISECONDS, true, () -> {
+ try {
+ StreamsGroup group = streamsGroup(groupId);
+ StreamsGroupMember member = group.getOrMaybeCreateMember(memberId, false);
+
+ if (member.memberEpoch() == memberEpoch) {
+ log.info("[GroupId {}] Member {} fenced from the group because " +
+ "it failed to transition from epoch {} within {}ms.",
+ groupId, memberId, memberEpoch, rebalanceTimeoutMs);
+
+ return streamsGroupFenceMember(group, member, null);
+ } else {
+ log.debug("[GroupId {}] Ignoring rebalance timeout for {} because the member " +
+ "left the epoch {}.", groupId, memberId, memberEpoch);
Review Comment:
Done. It's not in that epoch anymore, so progressed.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -3329,6 +3550,27 @@ private void scheduleShareGroupSessionTimeout(
);
}
+ /**
+ * Schedules (or reschedules) the session timeout for the member.
+ *
+ * @param groupId The group id.
+ * @param memberId The member id.
+ * @param sessionTimeoutMs The session timeout.
+ */
+ private void scheduleStreamsGroupSessionTimeout(
Review Comment:
Done
##########
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.
+ */
+ StreamsGroup getOrMaybeCreatePersistedStreamsGroup(
Review Comment:
Done
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorShardTest.java:
##########
@@ -830,6 +844,385 @@ public void testReplayConsumerGroupCurrentMemberAssignmentWithNullValue() {
verify(groupMetadataManager, times(1)).replay(key, null);
}
+ @Test
+ public void testReplayStreamsGroupMetadata() {
+ GroupMetadataManager groupMetadataManager = mock(GroupMetadataManager.class);
+ OffsetMetadataManager offsetMetadataManager = mock(OffsetMetadataManager.class);
+ CoordinatorMetrics coordinatorMetrics = mock(CoordinatorMetrics.class);
+ CoordinatorMetricsShard metricsShard = mock(CoordinatorMetricsShard.class);
+ GroupCoordinatorShard coordinator = new GroupCoordinatorShard(
+ new LogContext(),
+ groupMetadataManager,
+ offsetMetadataManager,
+ Time.SYSTEM,
+ new MockCoordinatorTimer<>(Time.SYSTEM),
+ mock(GroupCoordinatorConfig.class),
+ coordinatorMetrics,
+ metricsShard
+ );
+
+ StreamsGroupMetadataKey key = new StreamsGroupMetadataKey();
+ StreamsGroupMetadataValue value = new StreamsGroupMetadataValue();
+
+ coordinator.replay(0L, RecordBatch.NO_PRODUCER_ID, RecordBatch.NO_PRODUCER_EPOCH, CoordinatorRecord.record(
+ key,
+ new ApiMessageAndVersion(value, (short) 0)
+ ));
+
+ verify(groupMetadataManager, times(1)).replay(key, value);
Review Comment:
Done
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -687,6 +717,75 @@ ConsumerGroup getOrMaybeCreateConsumerGroup(
}
}
+ /**
+ * Gets or maybe creates a streams group without updating the groups map.
+ * The group will be materialized during the replay.
+ *
+ * @param groupId The group id.
+ * @param createIfNotExists A boolean indicating whether the group should be
+ * created if it does not exist or is an empty classic group.
+ *
+ * @return A StreamsGroup.
+ * @throws GroupIdNotFoundException if the group does not exist and createIfNotExists is false or
+ * if the group is not a streams group.
+ *
+ * Package private for testing.
+ */
+ StreamsGroup getOrMaybeCreateStreamsGroup(
+ String groupId,
+ boolean createIfNotExists
+ ) throws GroupIdNotFoundException {
+ Group group = groups.get(groupId);
+
+ if (group == null && !createIfNotExists) {
+ throw new GroupIdNotFoundException(String.format("Streams group %s not found.", groupId));
+ }
+
+ if (group == null) {
+ return new StreamsGroup(logContext, snapshotRegistry, groupId, metrics);
+ } else {
+ if (group.type() == STREAMS) {
+ return (StreamsGroup) group;
+ } else {
+ throw new GroupIdNotFoundException(String.format("Group %s is not a streams group.", groupId));
+ }
+ }
+ }
+
+ /**
+ * Gets a streams group by committed offset.
+ *
+ * @param groupId The group id.
+ * @param committedOffset A specified committed offset corresponding to this shard.
+ *
+ * @return A StreamsGroup.
+ * @throws GroupIdNotFoundException if the group does not exist or is not a streams group.
+ */
+ public StreamsGroup streamsGroup(
Review Comment:
Done
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -3138,6 +3274,45 @@ private void replaceMember(
));
}
+ /**
+ * Fences a member from a streams group and maybe downgrade the streams group to a classic group.
+ *
+ * @param group The group.
+ * @param member The member.
+ * @param response The response of the CoordinatorResult.
+ *
+ * @return The CoordinatorResult to be applied.
+ */
+ private <T> CoordinatorResult<T, CoordinatorRecord> streamsGroupFenceMember(
+ StreamsGroup group,
+ StreamsGroupMember member,
+ T response
+ ) {
+ List<CoordinatorRecord> records = new ArrayList<>();
+ removeStreamsMember(records, group.groupId(), member.memberId());
+
+ // We bump the group epoch.
+ int groupEpoch = group.groupEpoch() + 1;
+ records.add(newStreamsGroupEpochRecord(group.groupId(), groupEpoch));
+
+ cancelTimers(group.groupId(), member.memberId());
+
+ return new CoordinatorResult<>(records, response);
+ }
+
+ /**
+ * Write tombstones for the member. The order matters here.
+ *
+ * @param records The list of records to append the member assignment tombstone records.
+ * @param groupId The group id.
+ * @param memberId The member id.
+ */
+ private void removeStreamsMember(List<CoordinatorRecord> records, String groupId, String memberId) {
Review Comment:
Refactored, but it might become annoying in heartbeat when we assemble a larger collection of records.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -3962,7 +4359,167 @@ public void replay(
}
removeGroup(groupId);
}
+ }
+
+ /**
+ * Replays StreamsGroupMemberMetadataKey/Value to update the hard state of
+ * the streams group.
+ * It updates the subscription part of the member or deletes the member.
+ *
+ * @param key A StreamsGroupMemberMetadataKey key.
+ * @param value A StreamsGroupMemberMetadataValue record.
+ */
+ public void replay(
+ StreamsGroupMemberMetadataKey key,
+ StreamsGroupMemberMetadataValue value
+ ) {
+ String groupId = key.groupId();
+ String memberId = key.memberId();
+
+ StreamsGroup streamsGroup;
+ try {
+ streamsGroup = getOrMaybeCreatePersistedStreamsGroup(groupId, value != null);
+ } catch (GroupIdNotFoundException ex) {
+ // If the group does not exist and a tombstone is replayed, we can ignore it.
+ return;
+ }
+
+ if (value != null) {
+ StreamsGroupMember oldMember = streamsGroup.getOrMaybeCreateMember(memberId, true);
+ streamsGroup.updateMember(new StreamsGroupMember.Builder(oldMember)
+ .updateWith(value)
+ .build());
Review Comment:
I understand, but I don't think the pattern is too crazy. In the RPC handlers, we don't want to modify the internal state of the group, so we have "getters" and "updaters" and not operations that do both - that's why `getOrMaybeCreate` does not add the member to the group. In a sense, `getOrMaybeCreate` is very much like `getOrDefault`, just with a specific default implementation. That seems like a valid pattern to me, but I agree that one can be mislead into thinking that "creation" will also add the member to the streams group.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18730: Add replaying streams group state from offset topic [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #18809:
URL: https://github.com/apache/kafka/pull/18809#discussion_r1958380749
##########
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:
I was mostly following a change from the consumer groups here, but indeed it looks more lightweight
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18730: Add replaying streams group state from offset topic [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on PR #18809:
URL: https://github.com/apache/kafka/pull/18809#issuecomment-2663403671
> I noticed that the tests in the group coordinator do not use the following tag
>
> ```java
> @ExtendWith(MockitoExtension.class)
> public class XTest {
> ```
>
> Without `@ExtendWith(MockitoExtension.class)` mockito does not throw by default when a stubbing is not used and some other feature are not setup. See [javadoc.io/doc/org.mockito/mockito-junit-jupiter/latest/org/mockito/junit/jupiter/MockitoExtension.html](https://javadoc.io/doc/org.mockito/mockito-junit-jupiter/latest/org/mockito/junit/jupiter/MockitoExtension.html) Adding it also means, we need to add a library to the test dependency. In Streams we use that dependency in the tests. No big deal, IMO.
I haven't really dug into that, but it sounds like a useful feature to have. We can include this in our next PR
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18730: Add replaying streams group state from offset topic [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #18809:
URL: https://github.com/apache/kafka/pull/18809#discussion_r1949131601
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -3405,6 +3616,49 @@ private void scheduleConsumerGroupRebalanceTimeout(
});
}
+ /**
+ * Schedules a rebalance timeout for the member.
+ *
+ * @param groupId The group id.
+ * @param memberId The member id.
+ * @param memberEpoch The member epoch.
+ * @param rebalanceTimeoutMs The rebalance timeout.
+ */
+ private void scheduleStreamsGroupRebalanceTimeout(
+ String groupId,
+ String memberId,
+ int memberEpoch,
+ int rebalanceTimeoutMs
+ ) {
+ String key = streamsGroupRebalanceTimeoutKey(groupId, memberId);
+ timer.schedule(key, rebalanceTimeoutMs, TimeUnit.MILLISECONDS, true, () -> {
+ try {
+ StreamsGroup group = streamsGroup(groupId);
+ StreamsGroupMember member = group.getOrMaybeCreateMember(memberId, false);
+
+ if (member.memberEpoch() == memberEpoch) {
+ log.info("[GroupId {}] Member {} fenced from the group because " +
+ "it failed to transition from epoch {} within {}ms.",
+ groupId, memberId, memberEpoch, rebalanceTimeoutMs);
+
+ return streamsGroupFenceMember(group, member, null);
+ } else {
+ log.debug("[GroupId {}] Ignoring rebalance timeout for {} because the member " +
+ "left the epoch {}.", groupId, memberId, memberEpoch);
+ return new CoordinatorResult<>(Collections.emptyList());
+ }
+ } catch (GroupIdNotFoundException ex) {
+ log.debug("[GroupId {}] Could not fence {}} because the group does not exist.",
Review Comment:
Looking at the current implementation, I think timers aren't necessarily canceled when a group is deleted. We'd have to make sure that timers are always cancelled when a group is deleted. Would that be worth it?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -440,6 +468,7 @@ private GroupMetadataManager(
this.groupsByTopics = new TimelineHashMap<>(snapshotRegistry, 0);
this.groupConfigManager = groupConfigManager;
this.shareGroupAssignor = shareGroupAssignor;
+ this.streamsGroupSessionTimeoutMs = 45000;
Review Comment:
This is the default for consumer groups. We have a separate ticket to make it configurable, I just hardcoded this as a placeholder.
##########
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);
Review Comment:
Done. Note that this is part of the previous PR.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetrics.java:
##########
@@ -190,6 +202,55 @@ public GroupCoordinatorMetrics(MetricsRegistry registry, Metrics metrics) {
SHARE_GROUP_COUNT_STATE_TAG, ShareGroup.ShareGroupState.DEAD.toString()
);
+ streamsGroupCountMetricName = metrics.metricName(
+ GROUP_COUNT_METRIC_NAME,
+ METRICS_GROUP,
+ "The total number of groups using the streams rebalance protocol.",
+ Collections.singletonMap(GROUP_COUNT_PROTOCOL_TAG, Group.GroupType.STREAMS.toString())
+ );
+
+ streamsGroupCountEmptyMetricName = metrics.metricName(
+ STREAMS_GROUP_COUNT_METRIC_NAME,
+ METRICS_GROUP,
+ "The number of streams groups in empty state.",
+ Collections.singletonMap(STREAMS_GROUP_COUNT_STATE_TAG, StreamsGroupState.EMPTY.toString())
+ );
+
+ streamsGroupCountAssigningMetricName = metrics.metricName(
+ STREAMS_GROUP_COUNT_METRIC_NAME,
+ METRICS_GROUP,
+ "The number of streams groups in assigning state.",
+ Collections.singletonMap(STREAMS_GROUP_COUNT_STATE_TAG, StreamsGroupState.ASSIGNING.toString())
+ );
+
+ streamsGroupCountReconcilingMetricName = metrics.metricName(
+ STREAMS_GROUP_COUNT_METRIC_NAME,
+ METRICS_GROUP,
+ "The number of streams groups in reconciling state.",
+ Collections.singletonMap(STREAMS_GROUP_COUNT_STATE_TAG, StreamsGroupState.RECONCILING.toString())
+ );
+
+ streamsGroupCountStableMetricName = metrics.metricName(
+ STREAMS_GROUP_COUNT_METRIC_NAME,
+ METRICS_GROUP,
+ "The number of streams groups in stable state.",
+ Collections.singletonMap(STREAMS_GROUP_COUNT_STATE_TAG, StreamsGroupState.STABLE.toString())
+ );
+
+ streamsGroupCountDeadMetricName = metrics.metricName(
+ STREAMS_GROUP_COUNT_METRIC_NAME,
+ METRICS_GROUP,
+ "The number of streams groups in dead state.",
+ Collections.singletonMap(STREAMS_GROUP_COUNT_STATE_TAG, StreamsGroupState.DEAD.toString())
+ );
+
+ streamsGroupCountNotReadyMetricName = metrics.metricName(
+ STREAMS_GROUP_COUNT_METRIC_NAME,
+ METRICS_GROUP,
+ "The number of streams groups in not ready state.",
+ Collections.singletonMap(STREAMS_GROUP_COUNT_STATE_TAG, StreamsGroupState.NOT_READY.toString())
Review Comment:
I don't think individual groups are exposed in the current metrics. There are RPCs though to get the state of a specific state `DESCRIBE_GROUP` that can be used to implement such monitoring, I suppose.
##########
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));
Review Comment:
Done. Note that this is part of the previous PR.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18730: Add replaying streams group state from offset topic [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #18809:
URL: https://github.com/apache/kafka/pull/18809#discussion_r1952919827
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -15135,6 +15284,229 @@ public void testReplayConsumerGroupCurrentMemberAssignmentTombstone() {
assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.consumerGroup("bar"));
}
+ @Test
+ public void testReplayStreamsGroupMemberMetadata() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ StreamsGroupMember member = new StreamsGroupMember.Builder("member")
+ .setClientId("clientid")
+ .setClientHost("clienthost")
+ .setRackId("rackid")
+ .setInstanceId("instanceid")
+ .setRebalanceTimeoutMs(1000)
+ .setTopologyEpoch(10)
+ .setProcessId("processid")
+ .setUserEndpoint(new Endpoint().setHost("localhost").setPort(9999))
+ .setClientTags(Collections.singletonMap("key", "value"))
+ .build();
+
+ // The group and the member are created if they do not exist.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMemberRecord("foo", member));
+ assertEquals(member, context.groupMetadataManager.streamsGroup("foo").getOrMaybeCreateMember("member", false));
+ }
+
+ @Test
+ public void testReplayStreamsGroupMemberMetadataTombstone() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group still exists but the member is already gone. Replaying the
+ // StreamsGroupMemberMetadata tombstone should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupEpochRecord("foo", 10));
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMemberTombstoneRecord("foo", "m1"));
+ assertThrows(UnknownMemberIdException.class, () -> context.groupMetadataManager.streamsGroup("foo").getOrMaybeCreateMember("m1", false));
+
+ // The group may not exist at all. Replaying the StreamsGroupMemberMetadata tombstone
+ // should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMemberTombstoneRecord("bar", "m1"));
+ assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.streamsGroup("bar"));
+ }
+
+ @Test
+ public void testReplayStreamsGroupMetadata() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group is created if it does not exist.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupEpochRecord("foo", 10));
+ assertEquals(10, context.groupMetadataManager.streamsGroup("foo").groupEpoch());
+ }
+
+ @Test
+ public void testReplayStreamsGroupMetadataTombstone() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group may not exist at all. Replaying the StreamsGroupMetadata tombstone
+ // should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupEpochTombstoneRecord("foo"));
+ assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.streamsGroup("foo"));
+ }
+
+ @Test
+ public void testReplayStreamsGroupPartitionMetadata() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ Map<String, org.apache.kafka.coordinator.group.streams.TopicMetadata> metadata = Map.of(
+ "bar",
+ new org.apache.kafka.coordinator.group.streams.TopicMetadata(Uuid.randomUuid(), "bar", 10)
+ );
+
+ // The group is created if it does not exist.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupPartitionMetadataRecord("foo", metadata));
+ assertEquals(metadata, context.groupMetadataManager.streamsGroup("foo").partitionMetadata());
+ }
+
+ @Test
+ public void testReplayStreamsGroupPartitionMetadataTombstone() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group may not exist at all. Replaying the StreamsGroupPartitionMetadata tombstone
+ // should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupPartitionMetadataTombstoneRecord("foo"));
+ assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.streamsGroup("foo"));
+ }
+
+ @Test
+ public void testReplayStreamsGroupTargetAssignmentMember() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group is created if it does not exist.
+ final TasksTuple tasks =
+ new TasksTuple(
+ TaskAssignmentTestUtil.mkTasksPerSubtopology(
+ TaskAssignmentTestUtil.mkTasks("subtopology-1", 0, 1, 2)),
+ TaskAssignmentTestUtil.mkTasksPerSubtopology(
+ TaskAssignmentTestUtil.mkTasks("subtopology-1", 3, 4, 5)),
+ TaskAssignmentTestUtil.mkTasksPerSubtopology(
+ TaskAssignmentTestUtil.mkTasks("subtopology-1", 6, 7, 8))
+ );
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentRecord("foo", "m1", tasks));
+ assertEquals(tasks.activeTasks(), context.groupMetadataManager.streamsGroup("foo").targetAssignment("m1").activeTasks());
+ assertEquals(tasks.standbyTasks(), context.groupMetadataManager.streamsGroup("foo").targetAssignment("m1").standbyTasks());
+ assertEquals(tasks.warmupTasks(), context.groupMetadataManager.streamsGroup("foo").targetAssignment("m1").warmupTasks());
+ }
+
+ @Test
+ public void testReplayStreamsGroupTargetAssignmentMemberTombstone() {
Review Comment:
Done
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -15135,6 +15284,229 @@ public void testReplayConsumerGroupCurrentMemberAssignmentTombstone() {
assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.consumerGroup("bar"));
}
+ @Test
+ public void testReplayStreamsGroupMemberMetadata() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ StreamsGroupMember member = new StreamsGroupMember.Builder("member")
+ .setClientId("clientid")
+ .setClientHost("clienthost")
+ .setRackId("rackid")
+ .setInstanceId("instanceid")
+ .setRebalanceTimeoutMs(1000)
+ .setTopologyEpoch(10)
+ .setProcessId("processid")
+ .setUserEndpoint(new Endpoint().setHost("localhost").setPort(9999))
+ .setClientTags(Collections.singletonMap("key", "value"))
+ .build();
+
+ // The group and the member are created if they do not exist.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMemberRecord("foo", member));
+ assertEquals(member, context.groupMetadataManager.streamsGroup("foo").getOrMaybeCreateMember("member", false));
+ }
+
+ @Test
+ public void testReplayStreamsGroupMemberMetadataTombstone() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group still exists but the member is already gone. Replaying the
+ // StreamsGroupMemberMetadata tombstone should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupEpochRecord("foo", 10));
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMemberTombstoneRecord("foo", "m1"));
+ assertThrows(UnknownMemberIdException.class, () -> context.groupMetadataManager.streamsGroup("foo").getOrMaybeCreateMember("m1", false));
+
+ // The group may not exist at all. Replaying the StreamsGroupMemberMetadata tombstone
+ // should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMemberTombstoneRecord("bar", "m1"));
+ assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.streamsGroup("bar"));
+ }
+
+ @Test
+ public void testReplayStreamsGroupMetadata() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group is created if it does not exist.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupEpochRecord("foo", 10));
+ assertEquals(10, context.groupMetadataManager.streamsGroup("foo").groupEpoch());
+ }
+
+ @Test
+ public void testReplayStreamsGroupMetadataTombstone() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group may not exist at all. Replaying the StreamsGroupMetadata tombstone
+ // should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupEpochTombstoneRecord("foo"));
+ assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.streamsGroup("foo"));
+ }
+
+ @Test
+ public void testReplayStreamsGroupPartitionMetadata() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ Map<String, org.apache.kafka.coordinator.group.streams.TopicMetadata> metadata = Map.of(
+ "bar",
+ new org.apache.kafka.coordinator.group.streams.TopicMetadata(Uuid.randomUuid(), "bar", 10)
+ );
+
+ // The group is created if it does not exist.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupPartitionMetadataRecord("foo", metadata));
+ assertEquals(metadata, context.groupMetadataManager.streamsGroup("foo").partitionMetadata());
+ }
+
+ @Test
+ public void testReplayStreamsGroupPartitionMetadataTombstone() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group may not exist at all. Replaying the StreamsGroupPartitionMetadata tombstone
+ // should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupPartitionMetadataTombstoneRecord("foo"));
+ assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.streamsGroup("foo"));
+ }
+
+ @Test
+ public void testReplayStreamsGroupTargetAssignmentMember() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group is created if it does not exist.
+ final TasksTuple tasks =
+ new TasksTuple(
+ TaskAssignmentTestUtil.mkTasksPerSubtopology(
+ TaskAssignmentTestUtil.mkTasks("subtopology-1", 0, 1, 2)),
+ TaskAssignmentTestUtil.mkTasksPerSubtopology(
+ TaskAssignmentTestUtil.mkTasks("subtopology-1", 3, 4, 5)),
+ TaskAssignmentTestUtil.mkTasksPerSubtopology(
+ TaskAssignmentTestUtil.mkTasks("subtopology-1", 6, 7, 8))
+ );
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentRecord("foo", "m1", tasks));
+ assertEquals(tasks.activeTasks(), context.groupMetadataManager.streamsGroup("foo").targetAssignment("m1").activeTasks());
+ assertEquals(tasks.standbyTasks(), context.groupMetadataManager.streamsGroup("foo").targetAssignment("m1").standbyTasks());
+ assertEquals(tasks.warmupTasks(), context.groupMetadataManager.streamsGroup("foo").targetAssignment("m1").warmupTasks());
+ }
+
+ @Test
+ public void testReplayStreamsGroupTargetAssignmentMemberTombstone() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group may not exist at all. Replaying the StreamsGroupTargetAssignmentMember tombstone
+ // should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentTombstoneRecord("foo", "m1"));
+ assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.streamsGroup("foo"));
+ }
+
+ @Test
+ public void testReplayStreamsGroupTargetAssignmentMetadata() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group is created if it does not exist.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentEpochRecord("foo", 10));
+ assertEquals(10, context.groupMetadataManager.streamsGroup("foo").assignmentEpoch());
+ }
+
+ @Test
+ public void testReplayStreamsGroupTargetAssignmentMetadataTombstone() {
Review Comment:
Done
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -15135,6 +15284,229 @@ public void testReplayConsumerGroupCurrentMemberAssignmentTombstone() {
assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.consumerGroup("bar"));
}
+ @Test
+ public void testReplayStreamsGroupMemberMetadata() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ StreamsGroupMember member = new StreamsGroupMember.Builder("member")
+ .setClientId("clientid")
+ .setClientHost("clienthost")
+ .setRackId("rackid")
+ .setInstanceId("instanceid")
+ .setRebalanceTimeoutMs(1000)
+ .setTopologyEpoch(10)
+ .setProcessId("processid")
+ .setUserEndpoint(new Endpoint().setHost("localhost").setPort(9999))
+ .setClientTags(Collections.singletonMap("key", "value"))
+ .build();
+
+ // The group and the member are created if they do not exist.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMemberRecord("foo", member));
+ assertEquals(member, context.groupMetadataManager.streamsGroup("foo").getOrMaybeCreateMember("member", false));
+ }
+
+ @Test
+ public void testReplayStreamsGroupMemberMetadataTombstone() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group still exists but the member is already gone. Replaying the
+ // StreamsGroupMemberMetadata tombstone should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupEpochRecord("foo", 10));
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMemberTombstoneRecord("foo", "m1"));
+ assertThrows(UnknownMemberIdException.class, () -> context.groupMetadataManager.streamsGroup("foo").getOrMaybeCreateMember("m1", false));
+
+ // The group may not exist at all. Replaying the StreamsGroupMemberMetadata tombstone
+ // should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMemberTombstoneRecord("bar", "m1"));
+ assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.streamsGroup("bar"));
+ }
+
+ @Test
+ public void testReplayStreamsGroupMetadata() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group is created if it does not exist.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupEpochRecord("foo", 10));
+ assertEquals(10, context.groupMetadataManager.streamsGroup("foo").groupEpoch());
+ }
+
+ @Test
+ public void testReplayStreamsGroupMetadataTombstone() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group may not exist at all. Replaying the StreamsGroupMetadata tombstone
+ // should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupEpochTombstoneRecord("foo"));
+ assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.streamsGroup("foo"));
+ }
+
+ @Test
+ public void testReplayStreamsGroupPartitionMetadata() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ Map<String, org.apache.kafka.coordinator.group.streams.TopicMetadata> metadata = Map.of(
+ "bar",
+ new org.apache.kafka.coordinator.group.streams.TopicMetadata(Uuid.randomUuid(), "bar", 10)
+ );
+
+ // The group is created if it does not exist.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupPartitionMetadataRecord("foo", metadata));
+ assertEquals(metadata, context.groupMetadataManager.streamsGroup("foo").partitionMetadata());
+ }
+
+ @Test
+ public void testReplayStreamsGroupPartitionMetadataTombstone() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group may not exist at all. Replaying the StreamsGroupPartitionMetadata tombstone
+ // should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupPartitionMetadataTombstoneRecord("foo"));
+ assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.streamsGroup("foo"));
+ }
+
+ @Test
+ public void testReplayStreamsGroupTargetAssignmentMember() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group is created if it does not exist.
+ final TasksTuple tasks =
+ new TasksTuple(
+ TaskAssignmentTestUtil.mkTasksPerSubtopology(
+ TaskAssignmentTestUtil.mkTasks("subtopology-1", 0, 1, 2)),
+ TaskAssignmentTestUtil.mkTasksPerSubtopology(
+ TaskAssignmentTestUtil.mkTasks("subtopology-1", 3, 4, 5)),
+ TaskAssignmentTestUtil.mkTasksPerSubtopology(
+ TaskAssignmentTestUtil.mkTasks("subtopology-1", 6, 7, 8))
+ );
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentRecord("foo", "m1", tasks));
+ assertEquals(tasks.activeTasks(), context.groupMetadataManager.streamsGroup("foo").targetAssignment("m1").activeTasks());
+ assertEquals(tasks.standbyTasks(), context.groupMetadataManager.streamsGroup("foo").targetAssignment("m1").standbyTasks());
+ assertEquals(tasks.warmupTasks(), context.groupMetadataManager.streamsGroup("foo").targetAssignment("m1").warmupTasks());
+ }
+
+ @Test
+ public void testReplayStreamsGroupTargetAssignmentMemberTombstone() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group may not exist at all. Replaying the StreamsGroupTargetAssignmentMember tombstone
+ // should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentTombstoneRecord("foo", "m1"));
+ assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.streamsGroup("foo"));
+ }
+
+ @Test
+ public void testReplayStreamsGroupTargetAssignmentMetadata() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group is created if it does not exist.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentEpochRecord("foo", 10));
+ assertEquals(10, context.groupMetadataManager.streamsGroup("foo").assignmentEpoch());
+ }
+
+ @Test
+ public void testReplayStreamsGroupTargetAssignmentMetadataTombstone() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group may not exist at all. Replaying the StreamsGroupTargetAssignmentMetadata tombstone
+ // should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentEpochTombstoneRecord("foo"));
+ assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.streamsGroup("foo"));
+ }
+
+ @Test
+ public void testReplayStreamsGroupCurrentMemberAssignment() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ StreamsGroupMember member = new StreamsGroupMember.Builder("member")
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setState(org.apache.kafka.coordinator.group.streams.MemberState.UNRELEASED_TASKS)
+ .setAssignedTasks(new TasksTuple(
+ TaskAssignmentTestUtil.mkTasksPerSubtopology(TaskAssignmentTestUtil.mkTasks("subtopology-1", 0, 1, 2)),
+ TaskAssignmentTestUtil.mkTasksPerSubtopology(TaskAssignmentTestUtil.mkTasks("subtopology-1", 3, 4, 5)),
+ TaskAssignmentTestUtil.mkTasksPerSubtopology(TaskAssignmentTestUtil.mkTasks("subtopology-1", 6, 7, 8))
+ ))
+ .setTasksPendingRevocation(TasksTuple.EMPTY)
+ .build();
+
+ // The group and the member are created if they do not exist.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupCurrentAssignmentRecord("bar", member));
+ assertEquals(member, context.groupMetadataManager.streamsGroup("bar").getOrMaybeCreateMember("member", false));
+ }
+
+ @Test
+ public void testReplayStreamsGroupCurrentMemberAssignmentTombstone() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
Review Comment:
Done
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18730: Add replaying streams group state from offset topic [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #18809:
URL: https://github.com/apache/kafka/pull/18809#discussion_r1952918550
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -15135,6 +15284,229 @@ public void testReplayConsumerGroupCurrentMemberAssignmentTombstone() {
assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.consumerGroup("bar"));
}
+ @Test
+ public void testReplayStreamsGroupMemberMetadata() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ StreamsGroupMember member = new StreamsGroupMember.Builder("member")
+ .setClientId("clientid")
+ .setClientHost("clienthost")
+ .setRackId("rackid")
+ .setInstanceId("instanceid")
+ .setRebalanceTimeoutMs(1000)
+ .setTopologyEpoch(10)
+ .setProcessId("processid")
+ .setUserEndpoint(new Endpoint().setHost("localhost").setPort(9999))
+ .setClientTags(Collections.singletonMap("key", "value"))
+ .build();
+
+ // The group and the member are created if they do not exist.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMemberRecord("foo", member));
+ assertEquals(member, context.groupMetadataManager.streamsGroup("foo").getOrMaybeCreateMember("member", false));
+ }
+
+ @Test
+ public void testReplayStreamsGroupMemberMetadataTombstone() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group still exists but the member is already gone. Replaying the
+ // StreamsGroupMemberMetadata tombstone should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupEpochRecord("foo", 10));
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMemberTombstoneRecord("foo", "m1"));
+ assertThrows(UnknownMemberIdException.class, () -> context.groupMetadataManager.streamsGroup("foo").getOrMaybeCreateMember("m1", false));
+
+ // The group may not exist at all. Replaying the StreamsGroupMemberMetadata tombstone
+ // should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMemberTombstoneRecord("bar", "m1"));
+ assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.streamsGroup("bar"));
+ }
+
+ @Test
+ public void testReplayStreamsGroupMetadata() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group is created if it does not exist.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupEpochRecord("foo", 10));
+ assertEquals(10, context.groupMetadataManager.streamsGroup("foo").groupEpoch());
+ }
+
+ @Test
+ public void testReplayStreamsGroupMetadataTombstone() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group may not exist at all. Replaying the StreamsGroupMetadata tombstone
+ // should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupEpochTombstoneRecord("foo"));
+ assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.streamsGroup("foo"));
+ }
+
+ @Test
+ public void testReplayStreamsGroupPartitionMetadata() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ Map<String, org.apache.kafka.coordinator.group.streams.TopicMetadata> metadata = Map.of(
+ "bar",
+ new org.apache.kafka.coordinator.group.streams.TopicMetadata(Uuid.randomUuid(), "bar", 10)
+ );
+
+ // The group is created if it does not exist.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupPartitionMetadataRecord("foo", metadata));
+ assertEquals(metadata, context.groupMetadataManager.streamsGroup("foo").partitionMetadata());
+ }
+
+ @Test
+ public void testReplayStreamsGroupPartitionMetadataTombstone() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group may not exist at all. Replaying the StreamsGroupPartitionMetadata tombstone
+ // should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupPartitionMetadataTombstoneRecord("foo"));
+ assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.streamsGroup("foo"));
+ }
+
+ @Test
+ public void testReplayStreamsGroupTargetAssignmentMember() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group is created if it does not exist.
+ final TasksTuple tasks =
+ new TasksTuple(
+ TaskAssignmentTestUtil.mkTasksPerSubtopology(
+ TaskAssignmentTestUtil.mkTasks("subtopology-1", 0, 1, 2)),
+ TaskAssignmentTestUtil.mkTasksPerSubtopology(
+ TaskAssignmentTestUtil.mkTasks("subtopology-1", 3, 4, 5)),
+ TaskAssignmentTestUtil.mkTasksPerSubtopology(
+ TaskAssignmentTestUtil.mkTasks("subtopology-1", 6, 7, 8))
+ );
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentRecord("foo", "m1", tasks));
+ assertEquals(tasks.activeTasks(), context.groupMetadataManager.streamsGroup("foo").targetAssignment("m1").activeTasks());
+ assertEquals(tasks.standbyTasks(), context.groupMetadataManager.streamsGroup("foo").targetAssignment("m1").standbyTasks());
+ assertEquals(tasks.warmupTasks(), context.groupMetadataManager.streamsGroup("foo").targetAssignment("m1").warmupTasks());
+ }
+
+ @Test
+ public void testReplayStreamsGroupTargetAssignmentMemberTombstone() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group may not exist at all. Replaying the StreamsGroupTargetAssignmentMember tombstone
+ // should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentTombstoneRecord("foo", "m1"));
+ assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.streamsGroup("foo"));
+ }
+
+ @Test
+ public void testReplayStreamsGroupTargetAssignmentMetadata() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group is created if it does not exist.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentEpochRecord("foo", 10));
+ assertEquals(10, context.groupMetadataManager.streamsGroup("foo").assignmentEpoch());
+ }
+
+ @Test
+ public void testReplayStreamsGroupTargetAssignmentMetadataTombstone() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group may not exist at all. Replaying the StreamsGroupTargetAssignmentMetadata tombstone
+ // should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentEpochTombstoneRecord("foo"));
+ assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.streamsGroup("foo"));
+ }
+
+ @Test
+ public void testReplayStreamsGroupCurrentMemberAssignment() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ StreamsGroupMember member = new StreamsGroupMember.Builder("member")
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setState(org.apache.kafka.coordinator.group.streams.MemberState.UNRELEASED_TASKS)
+ .setAssignedTasks(new TasksTuple(
+ TaskAssignmentTestUtil.mkTasksPerSubtopology(TaskAssignmentTestUtil.mkTasks("subtopology-1", 0, 1, 2)),
+ TaskAssignmentTestUtil.mkTasksPerSubtopology(TaskAssignmentTestUtil.mkTasks("subtopology-1", 3, 4, 5)),
+ TaskAssignmentTestUtil.mkTasksPerSubtopology(TaskAssignmentTestUtil.mkTasks("subtopology-1", 6, 7, 8))
+ ))
+ .setTasksPendingRevocation(TasksTuple.EMPTY)
+ .build();
+
+ // The group and the member are created if they do not exist.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupCurrentAssignmentRecord("bar", member));
+ assertEquals(member, context.groupMetadataManager.streamsGroup("bar").getOrMaybeCreateMember("member", false));
+ }
+
+ @Test
+ public void testReplayStreamsGroupCurrentMemberAssignmentTombstone() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group still exists, but the member is already gone. Replaying the
+ // StreamsGroupCurrentMemberAssignment tombstone should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupEpochRecord("foo", 10));
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupCurrentAssignmentTombstoneRecord("foo", "m1"));
+ assertThrows(UnknownMemberIdException.class, () -> context.groupMetadataManager.streamsGroup("foo").getOrMaybeCreateMember("m1", false));
+
+ // The group may not exist at all. Replaying the StreamsGroupCurrentMemberAssignment tombstone
+ // should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupCurrentAssignmentTombstoneRecord("bar", "m1"));
+ assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.streamsGroup("bar"));
+ }
+
+ @Test
+ public void testReplayStreamsGroupTopology() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ StreamsGroupTopologyValue topology = new StreamsGroupTopologyValue()
+ .setEpoch(12)
+ .setSubtopologies(
+ List.of(
+ new StreamsGroupTopologyValue.Subtopology()
+ .setSubtopologyId("subtopology-1")
+ .setSourceTopics(List.of("source-topic"))
+ .setRepartitionSinkTopics(List.of("sink-topic"))
+ )
+ );
+
+ // The group and the topology are created if they do not exist.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupTopologyRecord("bar", topology));
+ final Optional<StreamsTopology> actualTopology = context.groupMetadataManager.streamsGroup("bar").topology();
+ assertTrue(actualTopology.isPresent(), "topology should be set");
+ assertEquals(topology.epoch(), actualTopology.get().topologyEpoch());
+ assertEquals(topology.subtopologies().size(), actualTopology.get().subtopologies().size());
+ assertEquals(
+ topology.subtopologies().iterator().next(),
+ actualTopology.get().subtopologies().values().iterator().next()
+ );
+ }
+
+ @Test
+ public void testReplayStreamsGroupTopologyTombstone() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
Review Comment:
Dome
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18730: Add replaying streams group state from offset topic [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck commented on code in PR #18809:
URL: https://github.com/apache/kafka/pull/18809#discussion_r1947288775
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetrics.java:
##########
@@ -190,6 +202,55 @@ public GroupCoordinatorMetrics(MetricsRegistry registry, Metrics metrics) {
SHARE_GROUP_COUNT_STATE_TAG, ShareGroup.ShareGroupState.DEAD.toString()
);
+ streamsGroupCountMetricName = metrics.metricName(
+ GROUP_COUNT_METRIC_NAME,
+ METRICS_GROUP,
+ "The total number of groups using the streams rebalance protocol.",
+ Collections.singletonMap(GROUP_COUNT_PROTOCOL_TAG, Group.GroupType.STREAMS.toString())
+ );
+
+ streamsGroupCountEmptyMetricName = metrics.metricName(
+ STREAMS_GROUP_COUNT_METRIC_NAME,
+ METRICS_GROUP,
+ "The number of streams groups in empty state.",
+ Collections.singletonMap(STREAMS_GROUP_COUNT_STATE_TAG, StreamsGroupState.EMPTY.toString())
+ );
+
+ streamsGroupCountAssigningMetricName = metrics.metricName(
+ STREAMS_GROUP_COUNT_METRIC_NAME,
+ METRICS_GROUP,
+ "The number of streams groups in assigning state.",
+ Collections.singletonMap(STREAMS_GROUP_COUNT_STATE_TAG, StreamsGroupState.ASSIGNING.toString())
+ );
+
+ streamsGroupCountReconcilingMetricName = metrics.metricName(
+ STREAMS_GROUP_COUNT_METRIC_NAME,
+ METRICS_GROUP,
+ "The number of streams groups in reconciling state.",
+ Collections.singletonMap(STREAMS_GROUP_COUNT_STATE_TAG, StreamsGroupState.RECONCILING.toString())
+ );
+
+ streamsGroupCountStableMetricName = metrics.metricName(
+ STREAMS_GROUP_COUNT_METRIC_NAME,
+ METRICS_GROUP,
+ "The number of streams groups in stable state.",
+ Collections.singletonMap(STREAMS_GROUP_COUNT_STATE_TAG, StreamsGroupState.STABLE.toString())
+ );
+
+ streamsGroupCountDeadMetricName = metrics.metricName(
+ STREAMS_GROUP_COUNT_METRIC_NAME,
+ METRICS_GROUP,
+ "The number of streams groups in dead state.",
+ Collections.singletonMap(STREAMS_GROUP_COUNT_STATE_TAG, StreamsGroupState.DEAD.toString())
+ );
+
+ streamsGroupCountNotReadyMetricName = metrics.metricName(
+ STREAMS_GROUP_COUNT_METRIC_NAME,
+ METRICS_GROUP,
+ "The number of streams groups in not ready state.",
+ Collections.singletonMap(STREAMS_GROUP_COUNT_STATE_TAG, StreamsGroupState.NOT_READY.toString())
Review Comment:
More of a meta question and not totally related to this code - but will we be able to drill down to individual groups in the different states via metrics?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -3405,6 +3616,49 @@ private void scheduleConsumerGroupRebalanceTimeout(
});
}
+ /**
+ * Schedules a rebalance timeout for the member.
+ *
+ * @param groupId The group id.
+ * @param memberId The member id.
+ * @param memberEpoch The member epoch.
+ * @param rebalanceTimeoutMs The rebalance timeout.
+ */
+ private void scheduleStreamsGroupRebalanceTimeout(
+ String groupId,
+ String memberId,
+ int memberEpoch,
+ int rebalanceTimeoutMs
+ ) {
+ String key = streamsGroupRebalanceTimeoutKey(groupId, memberId);
+ timer.schedule(key, rebalanceTimeoutMs, TimeUnit.MILLISECONDS, true, () -> {
+ try {
+ StreamsGroup group = streamsGroup(groupId);
+ StreamsGroupMember member = group.getOrMaybeCreateMember(memberId, false);
+
+ if (member.memberEpoch() == memberEpoch) {
+ log.info("[GroupId {}] Member {} fenced from the group because " +
+ "it failed to transition from epoch {} within {}ms.",
+ groupId, memberId, memberEpoch, rebalanceTimeoutMs);
+
+ return streamsGroupFenceMember(group, member, null);
+ } else {
+ log.debug("[GroupId {}] Ignoring rebalance timeout for {} because the member " +
+ "left the epoch {}.", groupId, memberId, memberEpoch);
+ return new CoordinatorResult<>(Collections.emptyList());
+ }
+ } catch (GroupIdNotFoundException ex) {
+ log.debug("[GroupId {}] Could not fence {}} because the group does not exist.",
Review Comment:
Why are these `debug` and not `error` or at least `warn`?
##########
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);
Review Comment:
possible to use `Optional` here and below instead?
##########
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));
Review Comment:
This will throw a NPE if the `topology` parameter is null, given the method above seems this is possible - maybe use `Optional.ofNullable` instead
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -440,6 +468,7 @@ private GroupMetadataManager(
this.groupsByTopics = new TimelineHashMap<>(snapshotRegistry, 0);
this.groupConfigManager = groupConfigManager;
this.shareGroupAssignor = shareGroupAssignor;
+ this.streamsGroupSessionTimeoutMs = 45000;
Review Comment:
Not that I have a problem with it,but how did we come to 45 sec?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18730: Add replaying streams group state from offset topic [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #18809:
URL: https://github.com/apache/kafka/pull/18809#discussion_r1954540074
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -3962,7 +4359,167 @@ public void replay(
}
removeGroup(groupId);
}
+ }
+
+ /**
+ * Replays StreamsGroupMemberMetadataKey/Value to update the hard state of
+ * the streams group.
+ * It updates the subscription part of the member or deletes the member.
+ *
+ * @param key A StreamsGroupMemberMetadataKey key.
+ * @param value A StreamsGroupMemberMetadataValue record.
+ */
+ public void replay(
+ StreamsGroupMemberMetadataKey key,
+ StreamsGroupMemberMetadataValue value
+ ) {
+ String groupId = key.groupId();
+ String memberId = key.memberId();
+
+ StreamsGroup streamsGroup;
+ try {
+ streamsGroup = getOrMaybeCreatePersistedStreamsGroup(groupId, value != null);
+ } catch (GroupIdNotFoundException ex) {
+ // If the group does not exist and a tombstone is replayed, we can ignore it.
+ return;
+ }
+
+ if (value != null) {
+ StreamsGroupMember oldMember = streamsGroup.getOrMaybeCreateMember(memberId, true);
+ streamsGroup.updateMember(new StreamsGroupMember.Builder(oldMember)
+ .updateWith(value)
+ .build());
Review Comment:
I understand. I was not advertizing to add the member to the group within `getOrMaybeCreate()`, but rather to explicitly build the member outside of `getOrMaybeCreate()`. We would then have a simple `get()` method and depeding on the result of the get use the `Builder` of the member. That would make it quite clear that the member is not added to the group. But maybe a renaming of `getOrMaybeCreate()` to `getOrDefault()` or similar would also make it clear.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18730: Add replaying streams group state from offset topic [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #18809:
URL: https://github.com/apache/kafka/pull/18809#discussion_r1958395986
##########
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:
getOrMaybeCreatePersistedStreamsGroup is called on the replay-path. That is, we call it while restoringn the internal state while replaying records from the offset topic.
getOrMaybeCreateStreamsGroup is called during RPC handlers, in particular StreamsGroupheartbeat, StreamsGroupDescribe, DeleteGroup
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18730: Add replaying streams group state from offset topic [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck commented on code in PR #18809:
URL: https://github.com/apache/kafka/pull/18809#discussion_r1953412766
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -3405,6 +3616,49 @@ private void scheduleConsumerGroupRebalanceTimeout(
});
}
+ /**
+ * Schedules a rebalance timeout for the member.
+ *
+ * @param groupId The group id.
+ * @param memberId The member id.
+ * @param memberEpoch The member epoch.
+ * @param rebalanceTimeoutMs The rebalance timeout.
+ */
+ private void scheduleStreamsGroupRebalanceTimeout(
+ String groupId,
+ String memberId,
+ int memberEpoch,
+ int rebalanceTimeoutMs
+ ) {
+ String key = streamsGroupRebalanceTimeoutKey(groupId, memberId);
+ timer.schedule(key, rebalanceTimeoutMs, TimeUnit.MILLISECONDS, true, () -> {
+ try {
+ StreamsGroup group = streamsGroup(groupId);
+ StreamsGroupMember member = group.getOrMaybeCreateMember(memberId, false);
+
+ if (member.memberEpoch() == memberEpoch) {
+ log.info("[GroupId {}] Member {} fenced from the group because " +
+ "it failed to transition from epoch {} within {}ms.",
+ groupId, memberId, memberEpoch, rebalanceTimeoutMs);
+
+ return streamsGroupFenceMember(group, member, null);
+ } else {
+ log.debug("[GroupId {}] Ignoring rebalance timeout for {} because the member " +
+ "left the epoch {}.", groupId, memberId, memberEpoch);
+ return new CoordinatorResult<>(Collections.emptyList());
+ }
+ } catch (GroupIdNotFoundException ex) {
+ log.debug("[GroupId {}] Could not fence {}} because the group does not exist.",
Review Comment:
SGTM
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18730: Add replaying streams group state from offset topic [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #18809:
URL: https://github.com/apache/kafka/pull/18809#discussion_r1952919339
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -15135,6 +15284,229 @@ public void testReplayConsumerGroupCurrentMemberAssignmentTombstone() {
assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.consumerGroup("bar"));
}
+ @Test
+ public void testReplayStreamsGroupMemberMetadata() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ StreamsGroupMember member = new StreamsGroupMember.Builder("member")
+ .setClientId("clientid")
+ .setClientHost("clienthost")
+ .setRackId("rackid")
+ .setInstanceId("instanceid")
+ .setRebalanceTimeoutMs(1000)
+ .setTopologyEpoch(10)
+ .setProcessId("processid")
+ .setUserEndpoint(new Endpoint().setHost("localhost").setPort(9999))
+ .setClientTags(Collections.singletonMap("key", "value"))
+ .build();
+
+ // The group and the member are created if they do not exist.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMemberRecord("foo", member));
+ assertEquals(member, context.groupMetadataManager.streamsGroup("foo").getOrMaybeCreateMember("member", false));
+ }
+
+ @Test
+ public void testReplayStreamsGroupMemberMetadataTombstone() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group still exists but the member is already gone. Replaying the
+ // StreamsGroupMemberMetadata tombstone should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupEpochRecord("foo", 10));
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMemberTombstoneRecord("foo", "m1"));
+ assertThrows(UnknownMemberIdException.class, () -> context.groupMetadataManager.streamsGroup("foo").getOrMaybeCreateMember("m1", false));
+
+ // The group may not exist at all. Replaying the StreamsGroupMemberMetadata tombstone
+ // should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMemberTombstoneRecord("bar", "m1"));
+ assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.streamsGroup("bar"));
+ }
+
+ @Test
+ public void testReplayStreamsGroupMetadata() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group is created if it does not exist.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupEpochRecord("foo", 10));
+ assertEquals(10, context.groupMetadataManager.streamsGroup("foo").groupEpoch());
+ }
+
+ @Test
+ public void testReplayStreamsGroupMetadataTombstone() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ // The group may not exist at all. Replaying the StreamsGroupMetadata tombstone
+ // should be a no-op.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupEpochTombstoneRecord("foo"));
+ assertThrows(GroupIdNotFoundException.class, () -> context.groupMetadataManager.streamsGroup("foo"));
+ }
+
+ @Test
+ public void testReplayStreamsGroupPartitionMetadata() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ Map<String, org.apache.kafka.coordinator.group.streams.TopicMetadata> metadata = Map.of(
+ "bar",
+ new org.apache.kafka.coordinator.group.streams.TopicMetadata(Uuid.randomUuid(), "bar", 10)
+ );
+
+ // The group is created if it does not exist.
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupPartitionMetadataRecord("foo", metadata));
+ assertEquals(metadata, context.groupMetadataManager.streamsGroup("foo").partitionMetadata());
+ }
+
+ @Test
+ public void testReplayStreamsGroupPartitionMetadataTombstone() {
Review Comment:
Done
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18730: Add replaying streams group state from offset topic [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #18809:
URL: https://github.com/apache/kafka/pull/18809#discussion_r1958385111
##########
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:
Note that this builder is just test code. I believe it's fine to allow null here - we may even want to construct tests with nulls, so I would not overly restrict this.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org