lucasbru commented on code in PR #19219:
URL: https://github.com/apache/kafka/pull/19219#discussion_r2028521537
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -18070,6 +18241,101 @@ public void testShareGroupDynamicConfigs() {
context.assertNoRebalanceTimeout(groupId, memberId);
}
+ @Test
+ public void testStreamsGroupDynamicConfigs() {
+ String groupId = "fooup";
+ String memberId = Uuid.randomUuid().toString();
+ String subtopology1 = "subtopology1";
+ String fooTopicName = "foo";
+ Uuid fooTopicId = Uuid.randomUuid();
+ Topology topology = new Topology().setSubtopologies(List.of(
+ new
Subtopology().setSubtopologyId(subtopology1).setSourceTopics(List.of(fooTopicName))
+ ));
+
+ MockTaskAssignor assignor = new MockTaskAssignor("sticky");
+ GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder()
+ .withStreamsGroupTaskAssignors(List.of(assignor))
+ .withMetadataImage(new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 6)
+ .addRacks()
+ .build())
+ .build();
+
+ assignor.prepareGroupAssignment(
+ Map.of(memberId,
TaskAssignmentTestUtil.mkTasksTuple(TaskRole.ACTIVE,
+ TaskAssignmentTestUtil.mkTasks(subtopology1, 0, 1, 2))));
+
+ // Session timer is scheduled on first heartbeat.
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord>
result =
+ context.streamsGroupHeartbeat(
+ new StreamsGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId)
+ .setMemberEpoch(0)
+ .setRebalanceTimeoutMs(10000)
+ .setTopology(topology)
+ .setActiveTasks(List.of())
+ .setStandbyTasks(List.of())
+ .setWarmupTasks(List.of()));
+ assertEquals(1, result.response().data().memberEpoch());
+ assertEquals(Map.of("num.standby.replicas", "0"),
assignor.lastPassedAssignmentConfigs());
+
+ // Verify heartbeat interval
+ assertEquals(5000, result.response().data().heartbeatIntervalMs());
+
+ // Verify that there is a session time.
+ context.assertSessionTimeout(groupId, memberId, 45000);
+
+ // Advance time.
+ assertEquals(
+ List.of(),
+ context.sleep(result.response().data().heartbeatIntervalMs())
+ );
+
+ // Dynamic update group config
+ Properties newGroupConfig = new Properties();
+ newGroupConfig.put(STREAMS_SESSION_TIMEOUT_MS_CONFIG, 50000);
+ newGroupConfig.put(STREAMS_HEARTBEAT_INTERVAL_MS_CONFIG, 10000);
+ newGroupConfig.put(STREAMS_NUM_STANDBY_REPLICAS_CONFIG, 2);
+ context.updateGroupConfig(groupId, newGroupConfig);
+
+ // Session timer is rescheduled on second heartbeat, new assignment
with new parameter is calculated.
+ result = context.streamsGroupHeartbeat(
+ new StreamsGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId)
+ .setMemberEpoch(result.response().data().memberEpoch())
+ .setRackId("bla"));
+ assertEquals(2, result.response().data().memberEpoch());
+
+ // Verify heartbeat interval
+ assertEquals(10000, result.response().data().heartbeatIntervalMs());
+
+ // Verify that there is a session time.
+ context.assertSessionTimeout(groupId, memberId, 50000);
+
+ // Verify that the new number of standby replicas is used
+ assertEquals(Map.of("num.standby.replicas", "2"),
assignor.lastPassedAssignmentConfigs());
+
+ // Advance time.
+ assertEquals(
+ List.of(),
+ context.sleep(result.response().data().heartbeatIntervalMs())
+ );
+
+ // Session timer is cancelled on leave.
Review Comment:
See above.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -18070,6 +18241,101 @@ public void testShareGroupDynamicConfigs() {
context.assertNoRebalanceTimeout(groupId, memberId);
}
+ @Test
+ public void testStreamsGroupDynamicConfigs() {
+ String groupId = "fooup";
+ String memberId = Uuid.randomUuid().toString();
+ String subtopology1 = "subtopology1";
+ String fooTopicName = "foo";
+ Uuid fooTopicId = Uuid.randomUuid();
+ Topology topology = new Topology().setSubtopologies(List.of(
+ new
Subtopology().setSubtopologyId(subtopology1).setSourceTopics(List.of(fooTopicName))
+ ));
+
+ MockTaskAssignor assignor = new MockTaskAssignor("sticky");
+ GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder()
+ .withStreamsGroupTaskAssignors(List.of(assignor))
+ .withMetadataImage(new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 6)
+ .addRacks()
+ .build())
+ .build();
+
+ assignor.prepareGroupAssignment(
+ Map.of(memberId,
TaskAssignmentTestUtil.mkTasksTuple(TaskRole.ACTIVE,
+ TaskAssignmentTestUtil.mkTasks(subtopology1, 0, 1, 2))));
+
+ // Session timer is scheduled on first heartbeat.
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord>
result =
+ context.streamsGroupHeartbeat(
+ new StreamsGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId)
+ .setMemberEpoch(0)
+ .setRebalanceTimeoutMs(10000)
+ .setTopology(topology)
+ .setActiveTasks(List.of())
+ .setStandbyTasks(List.of())
+ .setWarmupTasks(List.of()));
+ assertEquals(1, result.response().data().memberEpoch());
+ assertEquals(Map.of("num.standby.replicas", "0"),
assignor.lastPassedAssignmentConfigs());
+
+ // Verify heartbeat interval
+ assertEquals(5000, result.response().data().heartbeatIntervalMs());
+
+ // Verify that there is a session time.
+ context.assertSessionTimeout(groupId, memberId, 45000);
+
+ // Advance time.
+ assertEquals(
Review Comment:
This is just checking that no timers expire. In theory, one could refactor
this into a separate test. I copied this test from share group / consumer
groups, and I think it makes sense to keep it similar.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -18070,6 +18241,101 @@ public void testShareGroupDynamicConfigs() {
context.assertNoRebalanceTimeout(groupId, memberId);
}
+ @Test
+ public void testStreamsGroupDynamicConfigs() {
+ String groupId = "fooup";
+ String memberId = Uuid.randomUuid().toString();
+ String subtopology1 = "subtopology1";
+ String fooTopicName = "foo";
+ Uuid fooTopicId = Uuid.randomUuid();
+ Topology topology = new Topology().setSubtopologies(List.of(
+ new
Subtopology().setSubtopologyId(subtopology1).setSourceTopics(List.of(fooTopicName))
+ ));
+
+ MockTaskAssignor assignor = new MockTaskAssignor("sticky");
+ GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder()
+ .withStreamsGroupTaskAssignors(List.of(assignor))
+ .withMetadataImage(new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 6)
+ .addRacks()
+ .build())
+ .build();
+
+ assignor.prepareGroupAssignment(
+ Map.of(memberId,
TaskAssignmentTestUtil.mkTasksTuple(TaskRole.ACTIVE,
+ TaskAssignmentTestUtil.mkTasks(subtopology1, 0, 1, 2))));
+
+ // Session timer is scheduled on first heartbeat.
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord>
result =
+ context.streamsGroupHeartbeat(
+ new StreamsGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId)
+ .setMemberEpoch(0)
+ .setRebalanceTimeoutMs(10000)
+ .setTopology(topology)
+ .setActiveTasks(List.of())
+ .setStandbyTasks(List.of())
+ .setWarmupTasks(List.of()));
+ assertEquals(1, result.response().data().memberEpoch());
+ assertEquals(Map.of("num.standby.replicas", "0"),
assignor.lastPassedAssignmentConfigs());
+
+ // Verify heartbeat interval
+ assertEquals(5000, result.response().data().heartbeatIntervalMs());
+
+ // Verify that there is a session time.
+ context.assertSessionTimeout(groupId, memberId, 45000);
+
+ // Advance time.
+ assertEquals(
+ List.of(),
+ context.sleep(result.response().data().heartbeatIntervalMs())
+ );
+
+ // Dynamic update group config
+ Properties newGroupConfig = new Properties();
+ newGroupConfig.put(STREAMS_SESSION_TIMEOUT_MS_CONFIG, 50000);
+ newGroupConfig.put(STREAMS_HEARTBEAT_INTERVAL_MS_CONFIG, 10000);
+ newGroupConfig.put(STREAMS_NUM_STANDBY_REPLICAS_CONFIG, 2);
+ context.updateGroupConfig(groupId, newGroupConfig);
+
+ // Session timer is rescheduled on second heartbeat, new assignment
with new parameter is calculated.
+ result = context.streamsGroupHeartbeat(
+ new StreamsGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId)
+ .setMemberEpoch(result.response().data().memberEpoch())
+ .setRackId("bla"));
+ assertEquals(2, result.response().data().memberEpoch());
Review Comment:
Okay, this one I am fine with removing.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -18070,6 +18241,101 @@ public void testShareGroupDynamicConfigs() {
context.assertNoRebalanceTimeout(groupId, memberId);
}
+ @Test
+ public void testStreamsGroupDynamicConfigs() {
+ String groupId = "fooup";
+ String memberId = Uuid.randomUuid().toString();
+ String subtopology1 = "subtopology1";
+ String fooTopicName = "foo";
+ Uuid fooTopicId = Uuid.randomUuid();
+ Topology topology = new Topology().setSubtopologies(List.of(
+ new
Subtopology().setSubtopologyId(subtopology1).setSourceTopics(List.of(fooTopicName))
+ ));
+
+ MockTaskAssignor assignor = new MockTaskAssignor("sticky");
+ GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder()
+ .withStreamsGroupTaskAssignors(List.of(assignor))
+ .withMetadataImage(new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 6)
+ .addRacks()
+ .build())
+ .build();
+
+ assignor.prepareGroupAssignment(
+ Map.of(memberId,
TaskAssignmentTestUtil.mkTasksTuple(TaskRole.ACTIVE,
+ TaskAssignmentTestUtil.mkTasks(subtopology1, 0, 1, 2))));
+
+ // Session timer is scheduled on first heartbeat.
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord>
result =
+ context.streamsGroupHeartbeat(
+ new StreamsGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId)
+ .setMemberEpoch(0)
+ .setRebalanceTimeoutMs(10000)
+ .setTopology(topology)
+ .setActiveTasks(List.of())
+ .setStandbyTasks(List.of())
+ .setWarmupTasks(List.of()));
+ assertEquals(1, result.response().data().memberEpoch());
+ assertEquals(Map.of("num.standby.replicas", "0"),
assignor.lastPassedAssignmentConfigs());
+
+ // Verify heartbeat interval
+ assertEquals(5000, result.response().data().heartbeatIntervalMs());
+
+ // Verify that there is a session time.
+ context.assertSessionTimeout(groupId, memberId, 45000);
Review Comment:
Replaced by
`GroupCoordinatorConfig.STREAMS_GROUP_SESSION_TIMEOUT_MS_DEFAULT`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
