dongnuo123 commented on code in PR #15988: URL: https://github.com/apache/kafka/pull/15988#discussion_r1610735381
########## 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"), + null, + Collections.emptyList() + ) + ))) + ); + + // Member 1 has a member epoch smaller than the group epoch. + ConsumerGroupMember member1 = new ConsumerGroupMember.Builder(memberId1) + .setRebalanceTimeoutMs(rebalanceTimeout) + .setClassicMemberMetadata( + new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata() + .setSessionTimeoutMs(sessionTimeout) + .setSupportedProtocols(protocols) + ) + .setMemberEpoch(9) + .build(); + + // Member 2 has unrevoked partition. + ConsumerGroupMember member2 = new ConsumerGroupMember.Builder(memberId2) + .setState(MemberState.UNREVOKED_PARTITIONS) + .setRebalanceTimeoutMs(rebalanceTimeout) + .setPartitionsPendingRevocation(mkAssignment(mkTopicAssignment(fooTopicId, 0))) + .setClassicMemberMetadata( + new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata() + .setSessionTimeoutMs(sessionTimeout) + .setSupportedProtocols(protocols) + ) + .setMemberEpoch(10) + .build(); + + // Member 3 is in UNRELEASED_PARTITIONS and all the partitions in its target assignment are free. + ConsumerGroupMember member3 = new ConsumerGroupMember.Builder(memberId3) + .setState(MemberState.UNRELEASED_PARTITIONS) + .setRebalanceTimeoutMs(rebalanceTimeout) + .setAssignedPartitions(mkAssignment(mkTopicAssignment(barTopicId, 0))) + .setClassicMemberMetadata( + new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata() + .setSessionTimeoutMs(sessionTimeout) + .setSupportedProtocols(protocols) + ) + .setMemberEpoch(10) + .build(); + + GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder() + .withAssignors(Collections.singletonList(new MockPartitionAssignor("range"))) + .withConsumerGroup(new ConsumerGroupBuilder(groupId, 10) + .withMember(member1) + .withMember(member2) + .withMember(member3) + .withAssignment(memberId3, mkAssignment(mkTopicAssignment(barTopicId, 0, 1, 2)))) + .build(); + + Arrays.asList(memberId1, memberId2, memberId3).forEach(memberId -> { + CoordinatorResult<HeartbeatResponseData, CoordinatorRecord> heartbeatResult = context.sendClassicGroupHeartbeat( + new HeartbeatRequestData() + .setGroupId(groupId) + .setMemberId(memberId) + .setGenerationId(memberId.equals(memberId1) ? 9 : 10) + ); + assertEquals(Collections.emptyList(), heartbeatResult.records()); + assertEquals(Errors.REBALANCE_IN_PROGRESS.code(), heartbeatResult.response().errorCode()); + context.assertSessionTimeout(groupId, memberId, sessionTimeout); + context.assertJoinTimeout(groupId, memberId, rebalanceTimeout); + }); + } + + @Test + public void testClassicGroupHeartbeatToConsumerWithUnknownMember() { + String groupId = "group-id"; + + GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder() + .withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)) + .build(); + + assertThrows(UnknownMemberIdException.class, () -> context.sendClassicGroupHeartbeat( + new HeartbeatRequestData() + .setGroupId(groupId) + .setMemberId("unknown-member-id") + .setGenerationId(10) + )); + + assertThrows(UnknownMemberIdException.class, () -> context.sendClassicGroupHeartbeat( + new HeartbeatRequestData() + .setGroupId(groupId) + .setMemberId("unknown-member-id") + .setGroupInstanceId("unknown-instance-id") + .setGenerationId(10) + )); + } + + @Test + public void testClassicGroupHeartbeatToConsumerWithFencedInstanceId() { + String groupId = "group-id"; + String memberId = "member-id"; + String instanceId = "instance-id"; + + GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder() + .withConsumerGroup(new ConsumerGroupBuilder(groupId, 10) + .withMember(new ConsumerGroupMember.Builder(memberId) + .setInstanceId(instanceId) + .setMemberEpoch(10) + .setClassicMemberMetadata( + new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata() + .setSessionTimeoutMs(5000) + .setSupportedProtocols(Collections.emptyList()) + ) + .build())) + .build(); + + assertThrows(FencedInstanceIdException.class, () -> context.sendClassicGroupHeartbeat( + new HeartbeatRequestData() + .setGroupId(groupId) + .setMemberId("unknown-member-id") + .setGroupInstanceId(instanceId) + .setGenerationId(10) + )); + } + + @Test + public void testClassicGroupHeartbeatToConsumerWithIllegalGenerationId() { + String groupId = "group-id"; + String memberId = "member-id"; + + GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder() + .withConsumerGroup(new ConsumerGroupBuilder(groupId, 10) + .withMember(new ConsumerGroupMember.Builder(memberId) + .setMemberEpoch(10) + .setClassicMemberMetadata( + new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata() + .setSessionTimeoutMs(5000) + .setSupportedProtocols(Collections.emptyList()) + ) + .build())) + .build(); + + assertThrows(IllegalGenerationException.class, () -> context.sendClassicGroupHeartbeat( + new HeartbeatRequestData() + .setGroupId(groupId) + .setMemberId(memberId) + .setGenerationId(9) + )); + } + + @Test + public void testClassicGroupHeartbeatToConsumerWithMemberNotUsingClassicProtocol() { + String groupId = "group-id"; + String memberId = "member-id"; + + GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder() + .withConsumerGroup(new ConsumerGroupBuilder(groupId, 10) + .withMember(new ConsumerGroupMember.Builder(memberId) + .setMemberEpoch(10) + .build())) + .build(); + + assertThrows(UnknownMemberIdException.class, () -> context.sendClassicGroupHeartbeat( + new HeartbeatRequestData() + .setGroupId(groupId) + .setMemberId(memberId) + .setGenerationId(10) + )); + } + + @Test + public void testConsumerGroupMemberUsingClassicProtocolFencedWhenSessionTimeout() { + 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 session timeout + 1. + List<MockCoordinatorTimer.ExpiredTimeout<Void, CoordinatorRecord>> timeouts = context.sleep(sessionTimeout + 1); + + // The member is fenced from the group. + assertEquals(1, timeouts.size()); + MockCoordinatorTimer.ExpiredTimeout<Void, CoordinatorRecord> timeout = timeouts.get(0); + assertEquals(consumerGroupSessionTimeoutKey(groupId, memberId), timeout.key); + assertRecordsEquals( + Arrays.asList( + CoordinatorRecordHelpers.newCurrentAssignmentTombstoneRecord(groupId, memberId), + CoordinatorRecordHelpers.newTargetAssignmentTombstoneRecord(groupId, memberId), + CoordinatorRecordHelpers.newMemberSubscriptionTombstoneRecord(groupId, memberId), + CoordinatorRecordHelpers.newGroupEpochRecord(groupId, 11) + ), + timeout.result.records() + ); + } + + @Test + public void testConsumerGroupMemberUsingClassicProtocolFencedWhenJoinTimeout() { + String groupId = "group-id"; + String memberId = Uuid.randomUuid().toString(); + int rebalanceTimeout = 500; + + 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 whose member epoch is smaller than the group epoch. + GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder() + .withAssignors(Collections.singletonList(new MockPartitionAssignor("range"))) + .withConsumerGroup(new ConsumerGroupBuilder(groupId, 10) + .withMember(new ConsumerGroupMember.Builder(memberId) + .setRebalanceTimeoutMs(rebalanceTimeout) + .setClassicMemberMetadata( + new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata() + .setSessionTimeoutMs(5000) + .setSupportedProtocols(protocols) + ) + .setMemberEpoch(9) + .build())) + .build(); + + // Heartbeat to schedule the join timeout. + HeartbeatRequestData request = new HeartbeatRequestData() + .setGroupId(groupId) + .setMemberId(memberId) + .setGenerationId(9); + assertEquals( + Errors.REBALANCE_IN_PROGRESS.code(), + context.sendClassicGroupHeartbeat(request).response().errorCode() + ); + context.assertSessionTimeout(groupId, memberId, 5000); + context.assertJoinTimeout(groupId, memberId, rebalanceTimeout); + + // Advance clock by rebalance timeout + 1. + List<MockCoordinatorTimer.ExpiredTimeout<Void, CoordinatorRecord>> timeouts = context.sleep(rebalanceTimeout + 1); + + // The member is fenced from the group. + assertEquals(1, timeouts.size()); + MockCoordinatorTimer.ExpiredTimeout<Void, CoordinatorRecord> timeout = timeouts.get(0); + assertEquals(consumerGroupJoinKey(groupId, memberId), timeout.key); + assertRecordsEquals( + Arrays.asList( + CoordinatorRecordHelpers.newCurrentAssignmentTombstoneRecord(groupId, memberId), + CoordinatorRecordHelpers.newTargetAssignmentTombstoneRecord(groupId, memberId), + CoordinatorRecordHelpers.newMemberSubscriptionTombstoneRecord(groupId, memberId), Review Comment: Another possible way is to make groupMetadataManager#removeMember package private for testing. Not sure which one is better. https://github.com/apache/kafka/blob/27a6c156c49e375edea0e6f33a35c64c615db1b5/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java#L2049-L2053 -- 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