jeffkbkim commented on code in PR #15988:
URL: https://github.com/apache/kafka/pull/15988#discussion_r1610462585
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -12401,6 +12436,363 @@ public void
testClassicGroupSyncToConsumerGroupRebalanceInProgress() throws Exce
.withProtocolName("range")
.build())
);
+ context.assertJoinTimeout(groupId, memberId, 10000);
+ }
+
+ @Test
+ public void testClassicGroupHeartbeatToConsumerGroupMaintainsSession()
throws Exception {
+ String groupId = "group-id";
+ String memberId = Uuid.randomUuid().toString();
+ int sessionTimeout = 5000;
+
+ List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols =
Collections.singletonList(
+ new ConsumerGroupMemberMetadataValue.ClassicProtocol()
+ .setName("range")
+
.setMetadata(Utils.toArray(ConsumerProtocol.serializeSubscription(
+ new ConsumerPartitionAssignor.Subscription(
+ Arrays.asList("foo"),
+ null,
+ Collections.emptyList()
+ )
+ )))
+ );
+
+ // Consumer group with a member using the classic protocol.
+ GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder()
+ .withAssignors(Collections.singletonList(new
MockPartitionAssignor("range")))
+ .withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
+ .withMember(new ConsumerGroupMember.Builder(memberId)
+ .setClassicMemberMetadata(
+ new
ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
+ .setSessionTimeoutMs(sessionTimeout)
+ .setSupportedProtocols(protocols)
+ )
+ .setMemberEpoch(10)
+ .build()))
+ .build();
+
+ // Heartbeat to schedule the session timeout.
+ HeartbeatRequestData request = new HeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId)
+ .setGenerationId(10);
+ context.sendClassicGroupHeartbeat(request);
+ context.assertSessionTimeout(groupId, memberId, sessionTimeout);
+
+ // Advance clock by 1/2 of session timeout.
+
GroupMetadataManagerTestContext.assertNoOrEmptyResult(context.sleep(sessionTimeout
/ 2));
+
+ HeartbeatResponseData heartbeatResponse =
context.sendClassicGroupHeartbeat(request).response();
+ assertEquals(Errors.NONE.code(), heartbeatResponse.errorCode());
+ context.assertSessionTimeout(groupId, memberId, sessionTimeout);
+
+ // Advance clock by 1/2 of session timeout.
+
GroupMetadataManagerTestContext.assertNoOrEmptyResult(context.sleep(sessionTimeout
/ 2));
+
+ heartbeatResponse =
context.sendClassicGroupHeartbeat(request).response();
+ assertEquals(Errors.NONE.code(), heartbeatResponse.errorCode());
+ context.assertSessionTimeout(groupId, memberId, sessionTimeout);
+ }
+
+ @Test
+ public void testClassicGroupHeartbeatToConsumerGroupRebalanceInProgress()
throws Exception {
+ String groupId = "group-id";
+ String memberId1 = Uuid.randomUuid().toString();
+ String memberId2 = Uuid.randomUuid().toString();
+ String memberId3 = Uuid.randomUuid().toString();
+ Uuid fooTopicId = Uuid.randomUuid();
+ Uuid barTopicId = Uuid.randomUuid();
+ int sessionTimeout = 5000;
+ int rebalanceTimeout = 10000;
+
+ List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols =
Collections.singletonList(
+ new ConsumerGroupMemberMetadataValue.ClassicProtocol()
+ .setName("range")
+
.setMetadata(Utils.toArray(ConsumerProtocol.serializeSubscription(
+ new ConsumerPartitionAssignor.Subscription(
+ Arrays.asList("foo"),
Review Comment:
nit: Collections.singletonList("foo")
--
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
