MINOR: add group coordinator test coverage
Some edge cases are not currently being tested. Add more tests to cover.
Reviewers: Guozhang Wang
@@ -933,9 +933,28 @@ class GroupCoordinatorTest {
assertTrue(message.contains(followerInstanceId.get))
}
+ @Test
+ def staticMemberReJoinWithIllegalStateAsUnknownMember() {
+ staticMembersJoinAndRebalance(leaderInstanceId, followerInstanceId)
+ val group = groupCoordinator.groupManager.getGroup(groupId).get
+ group.transitionTo(PreparingRebalance)
+ group.transitionTo(Empty)
+
+ EasyMock.reset(replicaManager)
+
+ // Illegal state exception shall trigger since follower id resides in pending member bucket.
+ val expectedException = intercept[IllegalStateException] {
+ staticJoinGroup(groupId, JoinGroupRequest.UNKNOWN_MEMBER_ID, followerInstanceId, protocolType, protocolSuperset, clockAdvance = 1)
+ }
+
+ val message = expectedException.getMessage
+ assertTrue(message.contains(group.groupId))
+ assertTrue(message.contains(followerInstanceId.get))
+ }
+
@Test
def staticMemberReJoinWithIllegalArgumentAsMissingOldMember() {
- val _ = staticMembersJoinAndRebalance(leaderInstanceId, followerInstanceId)
+ staticMembersJoinAndRebalance(leaderInstanceId, followerInstanceId)
val group = groupCoordinator.groupManager.getGroup(groupId).get
val invalidMemberId = "invalid_member_id"
group.addStaticMember(followerInstanceId, invalidMemberId)
@@ -1185,6 +1204,21 @@ class GroupCoordinatorTest {
assertEquals(Errors.COORDINATOR_NOT_AVAILABLE, heartbeatResult)
}
+ @Test
+ def testheartbeatEmptyGroup() {
+ val memberId = "memberId"
+
+ val group = new GroupMetadata(groupId, Empty, new MockTime())
+ val member = new MemberMetadata(memberId, groupId, groupInstanceId,
+ ClientId, ClientHost, DefaultRebalanceTimeout, DefaultSessionTimeout,
+ protocolType, List(("range", Array.empty[Byte]), ("roundrobin", Array.empty[Byte])))
+
+ group.add(member)
+ groupCoordinator.groupManager.addGroup(group)
+ val heartbeatResult = heartbeat(groupId, memberId, 0)
+ assertEquals(Errors.UNKNOWN_MEMBER_ID, heartbeatResult)
+ }
+
@Test
def testHeartbeatUnknownConsumerExistingGroup() {
val memberId = JoinGroupRequest.UNKNOWN_MEMBER_ID
