[GitHub] [kafka] hachikuji commented on a diff in pull request #12910: MINOR: JoinGroup and SyncGroup callbacks should catch exceptions

[GitBox]

hachikuji commented on code in PR #12910:
URL: https://github.com/apache/kafka/pull/12910#discussion_r1033987364


##########
core/src/main/scala/kafka/coordinator/group/GroupMetadata.scala:
##########
@@ -546,9 +546,16 @@ private[group] class GroupMetadata(val groupId: String, 
initialState: GroupState
   def maybeInvokeJoinCallback(member: MemberMetadata,
                               joinGroupResult: JoinGroupResult): Unit = {
     if (member.isAwaitingJoin) {
-      member.awaitingJoinCallback(joinGroupResult)
-      member.awaitingJoinCallback = null
-      numMembersAwaitingJoin -= 1
+      try {
+        member.awaitingJoinCallback(joinGroupResult)
+      } catch {
+        case t: Throwable =>
+          error(s"Failed to invoke join callback for $member due to 
${t.getMessage}.")
+          member.awaitingJoinCallback(JoinGroupResult(member.memberId, 
Errors.UNKNOWN_SERVER_ERROR))

Review Comment:
   Yeah, I agree it is probably better to fail fast. We'd rather have an 
explicit error than see the consumer stuck in the rebalance. If we find cases 
that are transient in the future, we can refine the handling.



-- 
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