dajac commented on code in PR #12308:
URL: https://github.com/apache/kafka/pull/12308#discussion_r1015231438
##########
core/src/test/scala/unit/kafka/coordinator/group/GroupCoordinatorTest.scala:
##########
@@ -1036,6 +1036,52 @@ class GroupCoordinatorTest {
assertEquals(Errors.NONE, syncGroupWithOldMemberIdResult.error)
}
+ @Test
+ def
staticMemberRejoinWithUpdatedSessionAndRebalanceTimeoutsButCannotPersistChange():
Unit = {
+ val rebalanceResult = staticMembersJoinAndRebalance(leaderInstanceId,
followerInstanceId)
+ val joinGroupResult = staticJoinGroupWithPersistence(groupId,
JoinGroupRequest.UNKNOWN_MEMBER_ID, followerInstanceId, protocolType,
protocolSuperset, clockAdvance = 1, 2 * DefaultSessionTimeout, 2 *
DefaultRebalanceTimeout, appendRecordError = Errors.MESSAGE_TOO_LARGE)
+ checkJoinGroupResult(joinGroupResult,
+ Errors.UNKNOWN_SERVER_ERROR,
+ rebalanceResult.generation,
+ Set.empty,
+ Stable,
+ Some(protocolType))
+ assertTrue(groupCoordinator.groupManager.getGroup(groupId).isDefined)
+ val group = groupCoordinator.groupManager.getGroup(groupId).get
+ group.allMemberMetadata.foreach { member =>
+ assertEquals(member.sessionTimeoutMs, DefaultSessionTimeout)
+ assertEquals(member.rebalanceTimeoutMs, DefaultRebalanceTimeout)
+ }
+ }
+
+
+ @Test
+ def
staticMemberRejoinWithUpdatedSessionAndRebalanceTimeoutsAndPersistChange():
Unit = {
+ val rebalanceResult = staticMembersJoinAndRebalance(leaderInstanceId,
followerInstanceId)
+ val followerJoinGroupResult = staticJoinGroupWithPersistence(groupId,
JoinGroupRequest.UNKNOWN_MEMBER_ID, followerInstanceId, protocolType,
protocolSuperset, clockAdvance = 1, 2 * DefaultSessionTimeout, 2 *
DefaultRebalanceTimeout)
+ checkJoinGroupResult(followerJoinGroupResult,
+ Errors.NONE,
+ rebalanceResult.generation,
+ Set.empty,
+ Stable,
+ Some(protocolType))
+ val leaderJoinGroupResult = staticJoinGroupWithPersistence(groupId,
JoinGroupRequest.UNKNOWN_MEMBER_ID, leaderInstanceId, protocolType,
protocolSuperset, clockAdvance = 1, 2 * DefaultSessionTimeout, 2 *
DefaultRebalanceTimeout)
+ checkJoinGroupResult(leaderJoinGroupResult,
+ Errors.NONE,
+ rebalanceResult.generation,
+ Set[String](leaderInstanceId, followerInstanceId),
Review Comment:
nit: `[String]` is not necessary here. The scala compiler should be able to
infer this.
##########
core/src/test/scala/unit/kafka/coordinator/group/GroupCoordinatorTest.scala:
##########
@@ -1036,6 +1036,52 @@ class GroupCoordinatorTest {
assertEquals(Errors.NONE, syncGroupWithOldMemberIdResult.error)
}
+ @Test
+ def
staticMemberRejoinWithUpdatedSessionAndRebalanceTimeoutsButCannotPersistChange():
Unit = {
+ val rebalanceResult = staticMembersJoinAndRebalance(leaderInstanceId,
followerInstanceId)
+ val joinGroupResult = staticJoinGroupWithPersistence(groupId,
JoinGroupRequest.UNKNOWN_MEMBER_ID, followerInstanceId, protocolType,
protocolSuperset, clockAdvance = 1, 2 * DefaultSessionTimeout, 2 *
DefaultRebalanceTimeout, appendRecordError = Errors.MESSAGE_TOO_LARGE)
+ checkJoinGroupResult(joinGroupResult,
+ Errors.UNKNOWN_SERVER_ERROR,
+ rebalanceResult.generation,
+ Set.empty,
+ Stable,
+ Some(protocolType))
+ assertTrue(groupCoordinator.groupManager.getGroup(groupId).isDefined)
+ val group = groupCoordinator.groupManager.getGroup(groupId).get
+ group.allMemberMetadata.foreach { member =>
+ assertEquals(member.sessionTimeoutMs, DefaultSessionTimeout)
+ assertEquals(member.rebalanceTimeoutMs, DefaultRebalanceTimeout)
+ }
+ }
+
+
+ @Test
+ def
staticMemberRejoinWithUpdatedSessionAndRebalanceTimeoutsAndPersistChange():
Unit = {
+ val rebalanceResult = staticMembersJoinAndRebalance(leaderInstanceId,
followerInstanceId)
+ val followerJoinGroupResult = staticJoinGroupWithPersistence(groupId,
JoinGroupRequest.UNKNOWN_MEMBER_ID, followerInstanceId, protocolType,
protocolSuperset, clockAdvance = 1, 2 * DefaultSessionTimeout, 2 *
DefaultRebalanceTimeout)
+ checkJoinGroupResult(followerJoinGroupResult,
+ Errors.NONE,
+ rebalanceResult.generation,
+ Set.empty,
+ Stable,
+ Some(protocolType))
+ val leaderJoinGroupResult = staticJoinGroupWithPersistence(groupId,
JoinGroupRequest.UNKNOWN_MEMBER_ID, leaderInstanceId, protocolType,
protocolSuperset, clockAdvance = 1, 2 * DefaultSessionTimeout, 2 *
DefaultRebalanceTimeout)
+ checkJoinGroupResult(leaderJoinGroupResult,
+ Errors.NONE,
+ rebalanceResult.generation,
+ Set[String](leaderInstanceId, followerInstanceId),
+ Stable,
+ Some(protocolType),
+ leaderJoinGroupResult.leaderId,
+ leaderJoinGroupResult.memberId,
+ true)
+ assertTrue(groupCoordinator.groupManager.getGroup(groupId).isDefined)
+ val group = groupCoordinator.groupManager.getGroup(groupId).get
+ group.allMemberMetadata.foreach{ member =>
Review Comment:
nit: A space is missing before `{`.
--
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
