You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by da...@apache.org on 2024/09/14 16:30:37 UTC
(kafka) 01/01: MINOR: Log reason holding the completion of the join phase in the classic protocol
This is an automated email from the ASF dual-hosted git repository.
dajac pushed a commit to branch minor-add-logging-to-maybe-complete-join
in repository https://gitbox.apache.org/repos/asf/kafka.git
commit 9529bb33d2e70327969ea824307b44f12ec082a1
Author: David Jacot <dj...@confluent.io>
AuthorDate: Sat Sep 14 16:28:53 2024 +0000
MINOR: Log reason holding the completion of the join phase in the classic protocol
---
.../coordinator/group/GroupMetadataManager.java | 24 ++++++++++++++++------
1 file changed, 18 insertions(+), 6 deletions(-)
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
index 58d0070af3e..1a11d0c08da 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
@@ -4028,14 +4028,26 @@ public class GroupMetadataManager {
* @return The coordinator result that will be appended to the log.
*/
private CoordinatorResult<Void, CoordinatorRecord> maybeCompleteJoinPhase(ClassicGroup group) {
- if (group.isInState(PREPARING_REBALANCE) &&
- group.hasAllMembersJoined() &&
- group.previousState() != EMPTY
- ) {
- return completeClassicGroupJoin(group);
+ if (!group.isInState(PREPARING_REBALANCE)) {
+ log.debug("Cannot complete join phase of group {} because the group is in {} state.",
+ group.groupId(), group.currentState());
+ return EMPTY_RESULT;
}
- return EMPTY_RESULT;
+ if (group.previousState() == EMPTY) {
+ log.debug("Cannot complete join phase of group {} because this is an initial rebalance.",
+ group.groupId());
+ return EMPTY_RESULT;
+ }
+
+ if (!group.hasAllMembersJoined()) {
+ log.debug("Cannot complete join phase of group {} because not all the members have rejoined. " +
+ "Members={}, AwaitingJoinResponses={}, PendingJoinMembers={}.",
+ group.groupId(), group.numMembers(), group.numAwaitingJoinResponse(), group.numPendingJoinMembers());
+ return EMPTY_RESULT;
+ }
+
+ return completeClassicGroupJoin(group);
}
/**