You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2022/11/28 10:34:37 UTC
[GitHub] [kafka] dajac opened a new pull request, #12910: MINOR: JoinGroup and SyncGroup callbacks should catch exceptions
dajac opened a new pull request, #12910:
URL: https://github.com/apache/kafka/pull/12910
We recently had a bug causing the JoinGroup callback to thrown an exception (https://github.com/apache/kafka/pull/12909). When it happens, the exception is propagated to the caller and the JoinGroup callback is never completed. To make it worst, the member whose callback failed become a zombie because the group coordinator does not expire member with a pending callback.
This patch catch exceptions for both invocation of JoinGroup and SyncGroup callbacks and retry to complete them with a `UNKNOWN_SERVER_ERROR` error if they failed.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac merged pull request #12910: MINOR: JoinGroup and SyncGroup callbacks should catch exceptions
Posted by GitBox <gi...@apache.org>.
dajac merged PR #12910:
URL: https://github.com/apache/kafka/pull/12910
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a diff in pull request #12910: MINOR: JoinGroup and SyncGroup callbacks should catch exceptions
Posted by GitBox <gi...@apache.org>.
hachikuji commented on code in PR #12910:
URL: https://github.com/apache/kafka/pull/12910#discussion_r1033867871
##########
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:
I guess an alternative is to return a retriable error. We don't know if the failure was a transient issue.
##########
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}.")
Review Comment:
Perhaps we should include the full trace?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on a diff in pull request #12910: MINOR: JoinGroup and SyncGroup callbacks should catch exceptions
Posted by GitBox <gi...@apache.org>.
dajac commented on code in PR #12910:
URL: https://github.com/apache/kafka/pull/12910#discussion_r1033965897
##########
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 considered this. I tend to believe that errors on this path would be fatal errors anyway because they are/would be unexpected errors (likely due to bugs). UNKNOWN_SERVER_ERROR seems quite reasonable.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a diff in pull request #12910: MINOR: JoinGroup and SyncGroup callbacks should catch exceptions
Posted by GitBox <gi...@apache.org>.
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on pull request #12910: MINOR: JoinGroup and SyncGroup callbacks should catch exceptions
Posted by GitBox <gi...@apache.org>.
dajac commented on PR #12910:
URL: https://github.com/apache/kafka/pull/12910#issuecomment-1330291981
Merged to trunk and 3.3.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on a diff in pull request #12910: MINOR: JoinGroup and SyncGroup callbacks should catch exceptions
Posted by GitBox <gi...@apache.org>.
dajac commented on code in PR #12910:
URL: https://github.com/apache/kafka/pull/12910#discussion_r1033962116
##########
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}.")
Review Comment:
That makes sense.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org