DL1231 commented on code in PR #15067:
URL: https://github.com/apache/kafka/pull/15067#discussion_r1667622511
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -13426,6 +13429,99 @@ public void
testNoConversionWhenSizeExceedsClassicMaxGroupSize() throws Exceptio
assertEquals(Group.GroupType.CONSUMER,
context.groupMetadataManager.group(groupId).type());
}
+ @Test
+ public void testDynamicUpdateSessionTimeout() {
+ String groupId = "fooup";
+ // Use a static member id as it makes the test easier.
+ String memberId = Uuid.randomUuid().toString();
+
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+ GroupConfigManager configManager = createConfigManager();
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+ GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder()
+ .withConsumerGroupAssignors(Collections.singletonList(assignor))
+ .withMetadataImage(new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 6)
+ .addRacks()
+ .build())
+ .withGroupConfigManager(configManager)
+ .build();
+
+ assignor.prepareGroupAssignment(new GroupAssignment(
+ Collections.singletonMap(memberId, new
MemberAssignmentImpl(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5)
+ )))
+ ));
+
+ // Session timer is scheduled on first heartbeat.
+ CoordinatorResult<ConsumerGroupHeartbeatResponseData,
CoordinatorRecord> result =
+ context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId)
+ .setMemberEpoch(0)
+ .setRebalanceTimeoutMs(90000)
+ .setSubscribedTopicNames(Collections.singletonList("foo"))
+ .setTopicPartitions(Collections.emptyList()));
+ assertEquals(1, result.response().memberEpoch());
+
+ // Verify that there is a session time.
+ context.assertSessionTimeout(groupId, memberId, 45000);
+
+ // Advance time.
+ assertEquals(
+ Collections.emptyList(),
+ context.sleep(result.response().heartbeatIntervalMs())
+ );
+
+ // Dynamic update group config
+ Properties newGroupConfig = new Properties();
+ newGroupConfig.put(CONSUMER_SESSION_TIMEOUT_MS_CONFIG, 50000);
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
