showuon commented on a change in pull request #11566:
URL: https://github.com/apache/kafka/pull/11566#discussion_r771384033
##########
File path: core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala
##########
@@ -1249,7 +1260,7 @@ class GroupCoordinator(val brokerId: Int,
// for new members. If the new member is still there, we expect it to
retry.
completeAndScheduleNextExpiration(group, member, NewMemberJoinTimeoutMs)
- maybePrepareRebalance(group, s"Adding new member $memberId with group
instance id $groupInstanceId")
+ maybePrepareRebalance(group, s"Adding new member $memberId with group
instance id $groupInstanceId. Member joined due to $reason")
Review comment:
+1 for `; client reason: $reason`
##########
File path:
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
##########
@@ -467,6 +468,7 @@ boolean joinGroupIfNeeded(final Timer timer) {
final RuntimeException exception = future.exception();
resetJoinGroupFuture();
+ rejoinReason = "rebalance failed due to " +
exception.getClass() + " error: " + exception.getMessage();
Review comment:
+1 for `getSimpleName` for the class. In addition to the David's
suggestion, I think we should also remove the `due to`, because there is
already 1 `due to` in the sentence. ex: `rebalance failed: '$message' ($class)`
##########
File path:
clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java
##########
@@ -486,6 +487,43 @@ public void testRetainMemberIdAfterSyncGroupDisconnect() {
ensureActiveGroup(rejoinedGeneration, memberId);
}
+ @Test
+ public void testRejoinReason() {
+ setupCoordinator();
+
+ String memberId = "memberId";
+ int generation = 5;
+
+ // test initial reason
+ mockClient.prepareResponse(groupCoordinatorResponse(node,
Errors.NONE));
+ expectJoinGroup("", "initialized abstract coordinator", generation,
memberId);
+
+ // successful sync group response should reset reason
+ expectSyncGroup(generation, memberId);
+ ensureActiveGroup(generation, memberId);
+ assertEquals("", coordinator.rejoinReason());
+
+ // Force a rebalance
+ expectJoinGroup(memberId, "Manual test trigger", generation, memberId);
+ expectSyncGroup(generation, memberId);
+ coordinator.requestRejoin("Manual test trigger");
+ ensureActiveGroup(generation, memberId);
+ assertEquals("", coordinator.rejoinReason());
+
+ // max group size reached
+
mockClient.prepareResponse(joinGroupFollowerResponse(defaultGeneration,
memberId, JoinGroupRequest.UNKNOWN_MEMBER_ID, Errors.GROUP_MAX_SIZE_REACHED));
+ coordinator.requestRejoin("Manual test trigger 2");
+ try {
+ coordinator.joinGroupIfNeeded(mockTime.timer(100L));
+ } catch (GroupMaxSizeReachedException e) {
Review comment:
Actually, you can achieve what you want by `assertThrows` as below:
```java
Throwable e = assertThrows(GroupMaxSizeReachedException.class,
() -> coordinator.joinGroupIfNeeded(mockTime.timer(100L)));
// next join group request should contain exception message
expectJoinGroup(memberId, e.getMessage(), generation, memberId);
expectSyncGroup(generation, memberId);
ensureActiveGroup(generation, memberId);
assertEquals("", coordinator.rejoinReason());
```
Basically, the `assertThrows` is doing the similar thing as what you did
here (try/catch). It's recommended to use the `assertThrows` to understand the
exception is also a verification position in this test. Otherwise, let's say,
if someday, someone breaks the logic and makes the
`coordinator.joinGroupIfNeeded(mockTime.timer(100L));` works without exception
thrown, your current test can't catch this error, right?
--
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
