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 2021/06/10 13:41:50 UTC
[GitHub] [kafka] dajac opened a new pull request #10863: KAFKA-12890; Consumer group stuck in `CompletingRebalance`
dajac opened a new pull request #10863:
URL: https://github.com/apache/kafka/pull/10863
TODO
### 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on a change in pull request #10863: KAFKA-12890; Consumer group stuck in `CompletingRebalance`
Posted by GitBox <gi...@apache.org>.
dajac commented on a change in pull request #10863:
URL: https://github.com/apache/kafka/pull/10863#discussion_r652190661
##########
File path: core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala
##########
@@ -1450,7 +1457,95 @@ class GroupCoordinator(val brokerId: Int,
group.maybeInvokeJoinCallback(member, joinResult)
completeAndScheduleNextHeartbeatExpiration(group, member)
member.isNew = false
+
+ group.addPendingSyncMember(member.memberId)
}
+
+ schedulePendingSync(group)
+ }
+ }
+ }
+ }
+
+ private def removePendingSyncMember(
+ group: GroupMetadata,
+ memberId: String
+ ): Unit = {
+ group.removePendingSyncMember(memberId)
+ maybeCompleteSyncExpiration(group)
+ }
+
+ private def removeSyncExpiration(
+ group: GroupMetadata
+ ): Unit = {
+ group.clearPendingSyncMembers()
+ maybeCompleteSyncExpiration(group)
+ }
+
+ private def maybeCompleteSyncExpiration(
+ group: GroupMetadata
+ ): Unit = {
+ val groupKey = GroupKey(group.groupId)
+ syncPurgatory.checkAndComplete(groupKey)
+ }
+
+ private def schedulePendingSync(
+ group: GroupMetadata
+ ): Unit = {
+ val delayedSync = new DelayedSync(this, group, group.generationId, group.rebalanceTimeoutMs)
+ val groupKey = GroupKey(group.groupId)
+ syncPurgatory.tryCompleteElseWatch(delayedSync, Seq(groupKey))
+ }
+
+ def tryCompletePendingSync(
+ group: GroupMetadata,
+ generationId: Int,
+ forceComplete: () => Boolean
+ ): Boolean = {
+ group.inLock {
+ if (generationId != group.generationId) {
+ forceComplete()
+ } else {
+ group.currentState match {
+ case Dead | Empty | PreparingRebalance =>
+ forceComplete()
+ case CompletingRebalance | Stable =>
+ if (group.hasReceivedSyncFromAllMembers())
+ forceComplete()
+ else false
+ }
+ }
+ }
+ }
+
+ def onExpirePendingSync(
+ group: GroupMetadata,
+ generationId: Int
+ ): Unit = {
+ group.inLock {
+ if (generationId != group.generationId) {
+ debug(s"Received unexpected notification of sync expiration for ${group.groupId} " +
+ s" with an old generation $generationId.")
+ } else {
+ group.currentState match {
+ case Dead | Empty | PreparingRebalance =>
+ debug(s"Received unexpected notification of sync expiration after group ${group.groupId} " +
+ s"already transitioned to the ${group.currentState} state.")
+
+ case CompletingRebalance | Stable =>
+ if (!group.hasAllMembersJoined) {
Review comment:
Oops... It is definitely not. I introduced an predicate method in the group for this but use the wrong one here. My bad...
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #10863: KAFKA-12890; Consumer group stuck in `CompletingRebalance`
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #10863:
URL: https://github.com/apache/kafka/pull/10863#discussion_r652141670
##########
File path: core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala
##########
@@ -1450,7 +1457,95 @@ class GroupCoordinator(val brokerId: Int,
group.maybeInvokeJoinCallback(member, joinResult)
completeAndScheduleNextHeartbeatExpiration(group, member)
member.isNew = false
+
+ group.addPendingSyncMember(member.memberId)
}
+
+ schedulePendingSync(group)
+ }
+ }
+ }
+ }
+
+ private def removePendingSyncMember(
+ group: GroupMetadata,
+ memberId: String
+ ): Unit = {
+ group.removePendingSyncMember(memberId)
+ maybeCompleteSyncExpiration(group)
+ }
+
+ private def removeSyncExpiration(
Review comment:
One thing we seem to be missing is a call to `removeSyncExpiration` when the group is unloaded in `onGroupUnloaded`. That brings up an interesting limitation. When a group is loaded, there is no way to tell which followers have or have not sent the SyncGroup. The only reasonable thing to do is probably assume that they all did indeed send it since we could not rely on the request being resent if it had already been successfully received.
##########
File path: core/src/main/scala/kafka/coordinator/group/DelayedSync.scala
##########
@@ -0,0 +1,49 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.coordinator.group
+
+import kafka.server.DelayedOperation
+
+/**
+ * Delayed rebalance operation that is added to the purgatory when is completing the rebalance.
+ *
+ * Whenever a SyncGroup is received, checks that we received all the SyncGroup request from
+ * each member of the group; if yes, complete this operation.
+ *
+ * When the operation has expired, any known members that have not sent a SyncGroup requests
+ * are removed from the group. If any members is removed, the group is rebalanced.
+ */
+private[group] class DelayedSync(
Review comment:
There are a set of metrics that get exposed automatically for each purgatory we add. I'm considering if these will be useful or if we should add a flag to let us disable them.
##########
File path: core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala
##########
@@ -1450,7 +1457,95 @@ class GroupCoordinator(val brokerId: Int,
group.maybeInvokeJoinCallback(member, joinResult)
completeAndScheduleNextHeartbeatExpiration(group, member)
member.isNew = false
+
+ group.addPendingSyncMember(member.memberId)
}
+
+ schedulePendingSync(group)
+ }
+ }
+ }
+ }
+
+ private def removePendingSyncMember(
+ group: GroupMetadata,
+ memberId: String
+ ): Unit = {
+ group.removePendingSyncMember(memberId)
+ maybeCompleteSyncExpiration(group)
+ }
+
+ private def removeSyncExpiration(
+ group: GroupMetadata
+ ): Unit = {
+ group.clearPendingSyncMembers()
+ maybeCompleteSyncExpiration(group)
+ }
+
+ private def maybeCompleteSyncExpiration(
+ group: GroupMetadata
+ ): Unit = {
+ val groupKey = GroupKey(group.groupId)
+ syncPurgatory.checkAndComplete(groupKey)
+ }
+
+ private def schedulePendingSync(
+ group: GroupMetadata
+ ): Unit = {
+ val delayedSync = new DelayedSync(this, group, group.generationId, group.rebalanceTimeoutMs)
+ val groupKey = GroupKey(group.groupId)
+ syncPurgatory.tryCompleteElseWatch(delayedSync, Seq(groupKey))
+ }
+
+ def tryCompletePendingSync(
+ group: GroupMetadata,
+ generationId: Int,
+ forceComplete: () => Boolean
+ ): Boolean = {
+ group.inLock {
+ if (generationId != group.generationId) {
+ forceComplete()
+ } else {
+ group.currentState match {
+ case Dead | Empty | PreparingRebalance =>
+ forceComplete()
+ case CompletingRebalance | Stable =>
+ if (group.hasReceivedSyncFromAllMembers())
+ forceComplete()
+ else false
+ }
+ }
+ }
+ }
+
+ def onExpirePendingSync(
+ group: GroupMetadata,
+ generationId: Int
+ ): Unit = {
+ group.inLock {
+ if (generationId != group.generationId) {
+ debug(s"Received unexpected notification of sync expiration for ${group.groupId} " +
+ s" with an old generation $generationId.")
+ } else {
+ group.currentState match {
+ case Dead | Empty | PreparingRebalance =>
+ debug(s"Received unexpected notification of sync expiration after group ${group.groupId} " +
+ s"already transitioned to the ${group.currentState} state.")
+
+ case CompletingRebalance | Stable =>
+ if (!group.hasAllMembersJoined) {
Review comment:
Is this check right? Sync expiration begins after all members have joined. Did you mean to check the pending sync set?
##########
File path: core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala
##########
@@ -1450,7 +1457,95 @@ class GroupCoordinator(val brokerId: Int,
group.maybeInvokeJoinCallback(member, joinResult)
completeAndScheduleNextHeartbeatExpiration(group, member)
member.isNew = false
+
+ group.addPendingSyncMember(member.memberId)
}
+
+ schedulePendingSync(group)
+ }
+ }
+ }
+ }
+
+ private def removePendingSyncMember(
+ group: GroupMetadata,
+ memberId: String
+ ): Unit = {
+ group.removePendingSyncMember(memberId)
+ maybeCompleteSyncExpiration(group)
+ }
+
+ private def removeSyncExpiration(
+ group: GroupMetadata
+ ): Unit = {
+ group.clearPendingSyncMembers()
+ maybeCompleteSyncExpiration(group)
+ }
+
+ private def maybeCompleteSyncExpiration(
+ group: GroupMetadata
+ ): Unit = {
+ val groupKey = GroupKey(group.groupId)
+ syncPurgatory.checkAndComplete(groupKey)
+ }
+
+ private def schedulePendingSync(
+ group: GroupMetadata
+ ): Unit = {
+ val delayedSync = new DelayedSync(this, group, group.generationId, group.rebalanceTimeoutMs)
+ val groupKey = GroupKey(group.groupId)
+ syncPurgatory.tryCompleteElseWatch(delayedSync, Seq(groupKey))
+ }
+
+ def tryCompletePendingSync(
+ group: GroupMetadata,
+ generationId: Int,
+ forceComplete: () => Boolean
+ ): Boolean = {
+ group.inLock {
+ if (generationId != group.generationId) {
+ forceComplete()
+ } else {
+ group.currentState match {
+ case Dead | Empty | PreparingRebalance =>
+ forceComplete()
+ case CompletingRebalance | Stable =>
+ if (group.hasReceivedSyncFromAllMembers())
+ forceComplete()
+ else false
+ }
+ }
+ }
+ }
+
+ def onExpirePendingSync(
+ group: GroupMetadata,
+ generationId: Int
+ ): Unit = {
+ group.inLock {
+ if (generationId != group.generationId) {
+ debug(s"Received unexpected notification of sync expiration for ${group.groupId} " +
+ s" with an old generation $generationId.")
Review comment:
Might be useful to mention the current generation as well.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on pull request #10863: KAFKA-12890; Consumer group stuck in `CompletingRebalance`
Posted by GitBox <gi...@apache.org>.
dajac commented on pull request #10863:
URL: https://github.com/apache/kafka/pull/10863#issuecomment-860783985
@hachikuji We need to discuss wether we want to use the rebalance timeout or the session timeout here. It seems that the session timeout was used prior to https://github.com/apache/kafka/pull/8834 so we might want to do the same here to remain consistent. What do you think?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] showuon commented on a change in pull request #10863: KAFKA-12890; Consumer group stuck in `CompletingRebalance`
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #10863:
URL: https://github.com/apache/kafka/pull/10863#discussion_r649931726
##########
File path: core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala
##########
@@ -1450,12 +1463,67 @@ class GroupCoordinator(val brokerId: Int,
group.maybeInvokeJoinCallback(member, joinResult)
completeAndScheduleNextHeartbeatExpiration(group, member)
member.isNew = false
+
+ group.addPendingSyncMember(member.memberId)
}
+
+ val delayedSync = new DelayedSync(this, group, group.rebalanceTimeoutMs)
+ val groupKey = GroupKey(group.groupId)
+ syncPurgatory.tryCompleteElseWatch(delayedSync, Seq(groupKey))
}
}
}
}
+ private def maybeCompleteSyncExpiration(
+ group: GroupMetadata
+ ): Unit = {
+ val groupKey = GroupKey(group.groupId)
+ syncPurgatory.checkAndComplete(groupKey)
+ }
+
+ def tryCompletePendingSync(
+ group: GroupMetadata,
+ forceComplete: () => Boolean
+ ): Boolean = {
+ group.inLock {
+ group.currentState match {
+ case Dead | Empty | PreparingRebalance =>
+ forceComplete()
+ case CompletingRebalance | Stable =>
+ if (group.hasReceivedSyncFromAllMembers())
+ forceComplete()
+ else false
+ }
+ }
+ }
+
+ def onExpirePendingSync(
+ group: GroupMetadata
+ ): Unit = {
+ group.inLock {
+ group.currentState match {
+ case Dead | Empty | PreparingRebalance =>
+ debug(s"Received notification of sync expiration for group ${group.groupId} in ${group.currentState} state. Ignoring.")
Review comment:
nit: We could log this debug string (except `ignoring`) on the top of the `onExpirePendingSync` method, so that we can see the log in each situation.
##########
File path: core/src/main/scala/kafka/coordinator/group/DelayedSync.scala
##########
@@ -0,0 +1,50 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.coordinator.group
+
+import kafka.server.DelayedOperation
+
+/**
+ * Delayed rebalance operation that is added to the purgatory when is completing the rebalance.
+ *
+ * Whenever a SyncGroup is receives, checks that a SyncGroup has been received for each
+ * member of the group; if yes, complete this operation.
+ *
+ * When the operation has expired, any known members that have not sent a SyncGroup requests
+ * are removed from the group. If any members is removed, the group is rebalanced.
+ */
+private[group] class DelayedSync(
+ coordinator: GroupCoordinator,
+ group: GroupMetadata,
+ rebalanceTimeoutMs: Long
+) extends DelayedOperation(
+ rebalanceTimeoutMs,
+ Some(group.lock)
+) {
+ override def tryComplete(): Boolean = {
+ coordinator.tryCompletePendingSync(group, forceComplete _)
+ }
+
+ override def onExpiration(): Unit = {
+ coordinator.onExpirePendingSync(group)
+ }
+
+ override def onComplete(): Unit = {
+ // Nothing
Review comment:
nit: remove this comment.
##########
File path: core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala
##########
@@ -1209,7 +1216,11 @@ class GroupCoordinator(val brokerId: Int,
}
private def removeHeartbeatForLeavingMember(group: GroupMetadata, member: MemberMetadata): Unit = {
- val memberKey = MemberKey(group.groupId, member.memberId)
Review comment:
I don't think we should keep this overloading method here. We can just remove it and change all the caller to pass memberId (I searched, only 2 places called it)
##########
File path: core/src/main/scala/kafka/coordinator/group/DelayedSync.scala
##########
@@ -0,0 +1,50 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.coordinator.group
+
+import kafka.server.DelayedOperation
+
+/**
+ * Delayed rebalance operation that is added to the purgatory when is completing the rebalance.
+ *
+ * Whenever a SyncGroup is receives, checks that a SyncGroup has been received for each
+ * member of the group; if yes, complete this operation.
Review comment:
Are you trying to say:
Whenever a SyncGroup is receives, checks that **we received all the** SyncGroup **request from** each member of the group?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on pull request #10863: KAFKA-12890; Consumer group stuck in `CompletingRebalance`
Posted by GitBox <gi...@apache.org>.
dajac commented on pull request #10863:
URL: https://github.com/apache/kafka/pull/10863#issuecomment-860921205
@hachikuji Thanks for the clarification. Yes, that makes sense to me as well.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #10863: KAFKA-12890; Consumer group stuck in `CompletingRebalance`
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #10863:
URL: https://github.com/apache/kafka/pull/10863#discussion_r651224002
##########
File path: core/src/main/scala/kafka/coordinator/group/DelayedSync.scala
##########
@@ -0,0 +1,48 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.coordinator.group
+
+import kafka.server.DelayedOperation
+
+/**
+ * Delayed rebalance operation that is added to the purgatory when is completing the rebalance.
+ *
+ * Whenever a SyncGroup is receives, checks that we received all the SyncGroup request from
Review comment:
nit: Whenever a SyncGroup is receive**d**?
##########
File path: core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala
##########
@@ -1450,12 +1457,89 @@ class GroupCoordinator(val brokerId: Int,
group.maybeInvokeJoinCallback(member, joinResult)
completeAndScheduleNextHeartbeatExpiration(group, member)
member.isNew = false
+
+ group.addPendingSyncMember(member.memberId)
}
+
+ schedulePendingSync(group)
}
}
}
}
+ private def maybeRemovePendingSyncMember(
Review comment:
nit: `removePendingSyncMember` throws an exception if the member is not in the group, so does the "maybe" in the name make sense?
##########
File path: core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala
##########
@@ -1450,12 +1457,89 @@ class GroupCoordinator(val brokerId: Int,
group.maybeInvokeJoinCallback(member, joinResult)
completeAndScheduleNextHeartbeatExpiration(group, member)
member.isNew = false
+
+ group.addPendingSyncMember(member.memberId)
}
+
+ schedulePendingSync(group)
}
}
}
}
+ private def maybeRemovePendingSyncMember(
+ group: GroupMetadata,
+ memberId: String
+ ): Unit = {
+ group.removePendingSyncMember(memberId)
+ maybeCompleteSyncExpiration(group)
+ }
+
+ private def removeSyncExpiration(
+ group: GroupMetadata
+ ): Unit = {
+ group.clearPendingSyncMembers()
+ maybeCompleteSyncExpiration(group)
+ }
+
+ private def maybeCompleteSyncExpiration(
+ group: GroupMetadata
+ ): Unit = {
+ val groupKey = GroupKey(group.groupId)
+ syncPurgatory.checkAndComplete(groupKey)
+ }
+
+ private def schedulePendingSync(
+ group: GroupMetadata
+ ): Unit = {
+ val delayedSync = new DelayedSync(this, group, group.rebalanceTimeoutMs)
+ val groupKey = GroupKey(group.groupId)
+ syncPurgatory.tryCompleteElseWatch(delayedSync, Seq(groupKey))
+ }
+
+ def tryCompletePendingSync(
+ group: GroupMetadata,
+ forceComplete: () => Boolean
+ ): Boolean = {
+ group.inLock {
+ group.currentState match {
+ case Dead | Empty | PreparingRebalance =>
+ forceComplete()
+ case CompletingRebalance | Stable =>
+ if (group.hasReceivedSyncFromAllMembers())
+ forceComplete()
+ else false
+ }
+ }
+ }
+
+ def onExpirePendingSync(
+ group: GroupMetadata
+ ): Unit = {
+ group.inLock {
+ group.currentState match {
+ case Dead | Empty | PreparingRebalance =>
+ debug(s"Received unexpected notification of sync expiration after group ${group.groupId} " +
+ s"already transitioned to the ${group.currentState} state.")
+
+ case CompletingRebalance | Stable =>
+ if (!group.hasAllMembersJoined) {
+ val pendingSyncMembers = group.allPendingSyncMembers()
+
+ info(s"Group ${group.groupId} removed members who haven't " +
Review comment:
nit: this message might be redundant given the one in `prepareRebalance`.
##########
File path: core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala
##########
@@ -1450,12 +1457,89 @@ class GroupCoordinator(val brokerId: Int,
group.maybeInvokeJoinCallback(member, joinResult)
completeAndScheduleNextHeartbeatExpiration(group, member)
member.isNew = false
+
+ group.addPendingSyncMember(member.memberId)
}
+
+ schedulePendingSync(group)
}
}
}
}
+ private def maybeRemovePendingSyncMember(
+ group: GroupMetadata,
+ memberId: String
+ ): Unit = {
+ group.removePendingSyncMember(memberId)
+ maybeCompleteSyncExpiration(group)
+ }
+
+ private def removeSyncExpiration(
+ group: GroupMetadata
+ ): Unit = {
+ group.clearPendingSyncMembers()
+ maybeCompleteSyncExpiration(group)
+ }
+
+ private def maybeCompleteSyncExpiration(
+ group: GroupMetadata
+ ): Unit = {
+ val groupKey = GroupKey(group.groupId)
+ syncPurgatory.checkAndComplete(groupKey)
+ }
+
+ private def schedulePendingSync(
+ group: GroupMetadata
+ ): Unit = {
+ val delayedSync = new DelayedSync(this, group, group.rebalanceTimeoutMs)
Review comment:
I was considering if we should start the timer fresh after the group is joined. An alternative is to let the rebalance timeout cover the total time from when the rebalance is first triggered. On the other hand, if we are trying to allow for a full max.poll.interval.ms in between Join and Sync, then maybe we need to reset it.
##########
File path: core/src/main/scala/kafka/coordinator/group/DelayedSync.scala
##########
@@ -0,0 +1,48 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.coordinator.group
+
+import kafka.server.DelayedOperation
+
+/**
+ * Delayed rebalance operation that is added to the purgatory when is completing the rebalance.
+ *
+ * Whenever a SyncGroup is receives, checks that we received all the SyncGroup request from
+ * each member of the group; if yes, complete this operation.
+ *
+ * When the operation has expired, any known members that have not sent a SyncGroup requests
+ * are removed from the group. If any members is removed, the group is rebalanced.
+ */
+private[group] class DelayedSync(
+ coordinator: GroupCoordinator,
+ group: GroupMetadata,
Review comment:
Async operations always make me a little nervous. Would it be reasonable to add the generation to this as a field? That would give us a simple way to ensure that the delayed operation could not get applied to the wrong rebalance.
##########
File path: core/src/test/scala/unit/kafka/coordinator/group/GroupCoordinatorTest.scala
##########
@@ -2254,6 +2255,209 @@ class GroupCoordinatorTest {
assertEquals(0, group().numPending)
}
+ private def verifyHeartbeat(
+ joinGroupResult: JoinGroupResult,
+ expectedError: Errors
+ ): Unit = {
+ EasyMock.reset(replicaManager)
+ val heartbeatResult = heartbeat(
+ groupId,
+ joinGroupResult.memberId,
+ joinGroupResult.generationId
+ )
+ assertEquals(expectedError, heartbeatResult)
+ }
+
+ private def joinWithNMembers(nbMembers: Int): Seq[JoinGroupResult] = {
+ val requiredKnownMemberId = true
+
+ // First JoinRequests
+ var futures = 1.to(nbMembers).map { _ =>
+ EasyMock.reset(replicaManager)
+ sendJoinGroup(groupId, JoinGroupRequest.UNKNOWN_MEMBER_ID, protocolType, protocols,
+ None, DefaultSessionTimeout, DefaultRebalanceTimeout, requiredKnownMemberId)
+ }
+
+ // Get back the assigned member ids
+ val memberIds = futures.map(await(_, 1).memberId)
+
+ // Second JoinRequests
+ futures = memberIds.map { memberId =>
+ EasyMock.reset(replicaManager)
+ sendJoinGroup(groupId, memberId, protocolType, protocols,
+ None, DefaultSessionTimeout, DefaultRebalanceTimeout, requiredKnownMemberId)
+ }
+
+ timer.advanceClock(GroupInitialRebalanceDelay + 1)
+ timer.advanceClock(DefaultRebalanceTimeout + 1)
+
+ futures.map(await(_, 1))
+ }
+
+ @Test
+ def testRebalanceTimesOutWhenSyncRequestIsNotReceiced(): Unit = {
Review comment:
nit: typo `Receiced` (one more below as well)
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] showuon commented on a change in pull request #10863: KAFKA-12890; Consumer group stuck in `CompletingRebalance`
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #10863:
URL: https://github.com/apache/kafka/pull/10863#discussion_r650895147
##########
File path: core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala
##########
@@ -1450,12 +1463,67 @@ class GroupCoordinator(val brokerId: Int,
group.maybeInvokeJoinCallback(member, joinResult)
completeAndScheduleNextHeartbeatExpiration(group, member)
member.isNew = false
+
+ group.addPendingSyncMember(member.memberId)
}
+
+ val delayedSync = new DelayedSync(this, group, group.rebalanceTimeoutMs)
+ val groupKey = GroupKey(group.groupId)
+ syncPurgatory.tryCompleteElseWatch(delayedSync, Seq(groupKey))
}
}
}
}
+ private def maybeCompleteSyncExpiration(
+ group: GroupMetadata
+ ): Unit = {
+ val groupKey = GroupKey(group.groupId)
+ syncPurgatory.checkAndComplete(groupKey)
+ }
+
+ def tryCompletePendingSync(
+ group: GroupMetadata,
+ forceComplete: () => Boolean
+ ): Boolean = {
+ group.inLock {
+ group.currentState match {
+ case Dead | Empty | PreparingRebalance =>
+ forceComplete()
+ case CompletingRebalance | Stable =>
+ if (group.hasReceivedSyncFromAllMembers())
+ forceComplete()
+ else false
+ }
+ }
+ }
+
+ def onExpirePendingSync(
+ group: GroupMetadata
+ ): Unit = {
+ group.inLock {
+ group.currentState match {
+ case Dead | Empty | PreparingRebalance =>
+ debug(s"Received notification of sync expiration for group ${group.groupId} in ${group.currentState} state. Ignoring.")
Review comment:
SGTM. Thanks.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on a change in pull request #10863: KAFKA-12890; Consumer group stuck in `CompletingRebalance`
Posted by GitBox <gi...@apache.org>.
dajac commented on a change in pull request #10863:
URL: https://github.com/apache/kafka/pull/10863#discussion_r651729938
##########
File path: core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala
##########
@@ -1450,12 +1457,89 @@ class GroupCoordinator(val brokerId: Int,
group.maybeInvokeJoinCallback(member, joinResult)
completeAndScheduleNextHeartbeatExpiration(group, member)
member.isNew = false
+
+ group.addPendingSyncMember(member.memberId)
}
+
+ schedulePendingSync(group)
}
}
}
}
+ private def maybeRemovePendingSyncMember(
+ group: GroupMetadata,
+ memberId: String
+ ): Unit = {
+ group.removePendingSyncMember(memberId)
+ maybeCompleteSyncExpiration(group)
+ }
+
+ private def removeSyncExpiration(
+ group: GroupMetadata
+ ): Unit = {
+ group.clearPendingSyncMembers()
+ maybeCompleteSyncExpiration(group)
+ }
+
+ private def maybeCompleteSyncExpiration(
+ group: GroupMetadata
+ ): Unit = {
+ val groupKey = GroupKey(group.groupId)
+ syncPurgatory.checkAndComplete(groupKey)
+ }
+
+ private def schedulePendingSync(
+ group: GroupMetadata
+ ): Unit = {
+ val delayedSync = new DelayedSync(this, group, group.rebalanceTimeoutMs)
+ val groupKey = GroupKey(group.groupId)
+ syncPurgatory.tryCompleteElseWatch(delayedSync, Seq(groupKey))
+ }
+
+ def tryCompletePendingSync(
+ group: GroupMetadata,
+ forceComplete: () => Boolean
+ ): Boolean = {
+ group.inLock {
+ group.currentState match {
+ case Dead | Empty | PreparingRebalance =>
+ forceComplete()
+ case CompletingRebalance | Stable =>
+ if (group.hasReceivedSyncFromAllMembers())
+ forceComplete()
+ else false
+ }
+ }
+ }
+
+ def onExpirePendingSync(
+ group: GroupMetadata
+ ): Unit = {
+ group.inLock {
+ group.currentState match {
+ case Dead | Empty | PreparingRebalance =>
+ debug(s"Received unexpected notification of sync expiration after group ${group.groupId} " +
+ s"already transitioned to the ${group.currentState} state.")
+
+ case CompletingRebalance | Stable =>
+ if (!group.hasAllMembersJoined) {
+ val pendingSyncMembers = group.allPendingSyncMembers()
+
+ info(s"Group ${group.groupId} removed members who haven't " +
Review comment:
Good point. I will switch it to debug level. It is still useful for debugging.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on a change in pull request #10863: KAFKA-12890; Consumer group stuck in `CompletingRebalance`
Posted by GitBox <gi...@apache.org>.
dajac commented on a change in pull request #10863:
URL: https://github.com/apache/kafka/pull/10863#discussion_r650806388
##########
File path: core/src/main/scala/kafka/coordinator/group/DelayedSync.scala
##########
@@ -0,0 +1,50 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.coordinator.group
+
+import kafka.server.DelayedOperation
+
+/**
+ * Delayed rebalance operation that is added to the purgatory when is completing the rebalance.
+ *
+ * Whenever a SyncGroup is receives, checks that a SyncGroup has been received for each
+ * member of the group; if yes, complete this operation.
Review comment:
Right. I have reworded the sentence based on your comment.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on a change in pull request #10863: KAFKA-12890; Consumer group stuck in `CompletingRebalance`
Posted by GitBox <gi...@apache.org>.
dajac commented on a change in pull request #10863:
URL: https://github.com/apache/kafka/pull/10863#discussion_r650806015
##########
File path: core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala
##########
@@ -1450,12 +1463,67 @@ class GroupCoordinator(val brokerId: Int,
group.maybeInvokeJoinCallback(member, joinResult)
completeAndScheduleNextHeartbeatExpiration(group, member)
member.isNew = false
+
+ group.addPendingSyncMember(member.memberId)
}
+
+ val delayedSync = new DelayedSync(this, group, group.rebalanceTimeoutMs)
+ val groupKey = GroupKey(group.groupId)
+ syncPurgatory.tryCompleteElseWatch(delayedSync, Seq(groupKey))
}
}
}
}
+ private def maybeCompleteSyncExpiration(
+ group: GroupMetadata
+ ): Unit = {
+ val groupKey = GroupKey(group.groupId)
+ syncPurgatory.checkAndComplete(groupKey)
+ }
+
+ def tryCompletePendingSync(
+ group: GroupMetadata,
+ forceComplete: () => Boolean
+ ): Boolean = {
+ group.inLock {
+ group.currentState match {
+ case Dead | Empty | PreparingRebalance =>
+ forceComplete()
+ case CompletingRebalance | Stable =>
+ if (group.hasReceivedSyncFromAllMembers())
+ forceComplete()
+ else false
+ }
+ }
+ }
+
+ def onExpirePendingSync(
+ group: GroupMetadata
+ ): Unit = {
+ group.inLock {
+ group.currentState match {
+ case Dead | Empty | PreparingRebalance =>
+ debug(s"Received notification of sync expiration for group ${group.groupId} in ${group.currentState} state. Ignoring.")
Review comment:
I am not really convinced by this. My intent was to have a comment for the unexpected cases. I prefer to keep it as it is.
##########
File path: core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala
##########
@@ -1209,7 +1216,11 @@ class GroupCoordinator(val brokerId: Int,
}
private def removeHeartbeatForLeavingMember(group: GroupMetadata, member: MemberMetadata): Unit = {
- val memberKey = MemberKey(group.groupId, member.memberId)
Review comment:
Make 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on pull request #10863: KAFKA-12890; Consumer group stuck in `CompletingRebalance`
Posted by GitBox <gi...@apache.org>.
dajac commented on pull request #10863:
URL: https://github.com/apache/kafka/pull/10863#issuecomment-861594180
@hachikuji Thanks for your comments. I have addressed them.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on pull request #10863: KAFKA-12890; Consumer group stuck in `CompletingRebalance`
Posted by GitBox <gi...@apache.org>.
dajac commented on pull request #10863:
URL: https://github.com/apache/kafka/pull/10863#issuecomment-860556106
@showuon Thanks for your comments. I have addressed them.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on a change in pull request #10863: KAFKA-12890; Consumer group stuck in `CompletingRebalance`
Posted by GitBox <gi...@apache.org>.
dajac commented on a change in pull request #10863:
URL: https://github.com/apache/kafka/pull/10863#discussion_r652188256
##########
File path: core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala
##########
@@ -1450,7 +1457,95 @@ class GroupCoordinator(val brokerId: Int,
group.maybeInvokeJoinCallback(member, joinResult)
completeAndScheduleNextHeartbeatExpiration(group, member)
member.isNew = false
+
+ group.addPendingSyncMember(member.memberId)
}
+
+ schedulePendingSync(group)
+ }
+ }
+ }
+ }
+
+ private def removePendingSyncMember(
+ group: GroupMetadata,
+ memberId: String
+ ): Unit = {
+ group.removePendingSyncMember(memberId)
+ maybeCompleteSyncExpiration(group)
+ }
+
+ private def removeSyncExpiration(
Review comment:
Definitely. I will add that.
I cannot really think of any other doable alternatives. At least, we know that the leader has sent the SyncGroup as the group transitioned to stable.
Throwing few ideas:
* Could we store the pending set in the log? That does not seem really doable as the set could be updated before the write is acknowledged. Therefore we could think that a member is pending after a failover but it might not be.
* We could transition to Stable only when all pending members have joined. We already discussed this and agreed that this is not ideal as it would delay processing.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on pull request #10863: KAFKA-12890; Consumer group stuck in `CompletingRebalance`
Posted by GitBox <gi...@apache.org>.
hachikuji commented on pull request #10863:
URL: https://github.com/apache/kafka/pull/10863#issuecomment-860914096
@dajac Given the background from KAFKA-10134, the rebalance timeout seems like the right one in my mind. We're basically allowing for a delay between the JoinGroup and SyncGroup caused by consumer processing while the rebalance is in progress. This time is bounded by max.poll.interval.ms which is used as the rebalance timeout internally. Does that make 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on a change in pull request #10863: KAFKA-12890; Consumer group stuck in `CompletingRebalance`
Posted by GitBox <gi...@apache.org>.
dajac commented on a change in pull request #10863:
URL: https://github.com/apache/kafka/pull/10863#discussion_r652190153
##########
File path: core/src/main/scala/kafka/coordinator/group/DelayedSync.scala
##########
@@ -0,0 +1,49 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.coordinator.group
+
+import kafka.server.DelayedOperation
+
+/**
+ * Delayed rebalance operation that is added to the purgatory when is completing the rebalance.
+ *
+ * Whenever a SyncGroup is received, checks that we received all the SyncGroup request from
+ * each member of the group; if yes, complete this operation.
+ *
+ * When the operation has expired, any known members that have not sent a SyncGroup requests
+ * are removed from the group. If any members is removed, the group is rebalanced.
+ */
+private[group] class DelayedSync(
Review comment:
Ah.. I was not aware of this. This means that I cannot change the purgatory name like I did.
I think that it is useful to have some metrics. The issue is that we would need a KIP, right?
As an alternative, we could perhaps share the purgatory for the delayed join and the delayed sync. We could use different keys for instance. What do you think?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on a change in pull request #10863: KAFKA-12890; Consumer group stuck in `CompletingRebalance`
Posted by GitBox <gi...@apache.org>.
dajac commented on a change in pull request #10863:
URL: https://github.com/apache/kafka/pull/10863#discussion_r652188355
##########
File path: core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala
##########
@@ -1450,7 +1457,95 @@ class GroupCoordinator(val brokerId: Int,
group.maybeInvokeJoinCallback(member, joinResult)
completeAndScheduleNextHeartbeatExpiration(group, member)
member.isNew = false
+
+ group.addPendingSyncMember(member.memberId)
}
+
+ schedulePendingSync(group)
+ }
+ }
+ }
+ }
+
+ private def removePendingSyncMember(
+ group: GroupMetadata,
+ memberId: String
+ ): Unit = {
+ group.removePendingSyncMember(memberId)
+ maybeCompleteSyncExpiration(group)
+ }
+
+ private def removeSyncExpiration(
+ group: GroupMetadata
+ ): Unit = {
+ group.clearPendingSyncMembers()
+ maybeCompleteSyncExpiration(group)
+ }
+
+ private def maybeCompleteSyncExpiration(
+ group: GroupMetadata
+ ): Unit = {
+ val groupKey = GroupKey(group.groupId)
+ syncPurgatory.checkAndComplete(groupKey)
+ }
+
+ private def schedulePendingSync(
+ group: GroupMetadata
+ ): Unit = {
+ val delayedSync = new DelayedSync(this, group, group.generationId, group.rebalanceTimeoutMs)
+ val groupKey = GroupKey(group.groupId)
+ syncPurgatory.tryCompleteElseWatch(delayedSync, Seq(groupKey))
+ }
+
+ def tryCompletePendingSync(
+ group: GroupMetadata,
+ generationId: Int,
+ forceComplete: () => Boolean
+ ): Boolean = {
+ group.inLock {
+ if (generationId != group.generationId) {
+ forceComplete()
+ } else {
+ group.currentState match {
+ case Dead | Empty | PreparingRebalance =>
+ forceComplete()
+ case CompletingRebalance | Stable =>
+ if (group.hasReceivedSyncFromAllMembers())
+ forceComplete()
+ else false
+ }
+ }
+ }
+ }
+
+ def onExpirePendingSync(
+ group: GroupMetadata,
+ generationId: Int
+ ): Unit = {
+ group.inLock {
+ if (generationId != group.generationId) {
+ debug(s"Received unexpected notification of sync expiration for ${group.groupId} " +
+ s" with an old generation $generationId.")
Review comment:
Will do.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on a change in pull request #10863: KAFKA-12890; Consumer group stuck in `CompletingRebalance`
Posted by GitBox <gi...@apache.org>.
dajac commented on a change in pull request #10863:
URL: https://github.com/apache/kafka/pull/10863#discussion_r651728393
##########
File path: core/src/main/scala/kafka/coordinator/group/DelayedSync.scala
##########
@@ -0,0 +1,48 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.coordinator.group
+
+import kafka.server.DelayedOperation
+
+/**
+ * Delayed rebalance operation that is added to the purgatory when is completing the rebalance.
+ *
+ * Whenever a SyncGroup is receives, checks that we received all the SyncGroup request from
+ * each member of the group; if yes, complete this operation.
+ *
+ * When the operation has expired, any known members that have not sent a SyncGroup requests
+ * are removed from the group. If any members is removed, the group is rebalanced.
+ */
+private[group] class DelayedSync(
+ coordinator: GroupCoordinator,
+ group: GroupMetadata,
Review comment:
Yeah, that's a good idea. Let me do this.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on a change in pull request #10863: KAFKA-12890; Consumer group stuck in `CompletingRebalance`
Posted by GitBox <gi...@apache.org>.
dajac commented on a change in pull request #10863:
URL: https://github.com/apache/kafka/pull/10863#discussion_r651765967
##########
File path: core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala
##########
@@ -1450,12 +1457,89 @@ class GroupCoordinator(val brokerId: Int,
group.maybeInvokeJoinCallback(member, joinResult)
completeAndScheduleNextHeartbeatExpiration(group, member)
member.isNew = false
+
+ group.addPendingSyncMember(member.memberId)
}
+
+ schedulePendingSync(group)
}
}
}
}
+ private def maybeRemovePendingSyncMember(
+ group: GroupMetadata,
+ memberId: String
+ ): Unit = {
+ group.removePendingSyncMember(memberId)
+ maybeCompleteSyncExpiration(group)
+ }
+
+ private def removeSyncExpiration(
+ group: GroupMetadata
+ ): Unit = {
+ group.clearPendingSyncMembers()
+ maybeCompleteSyncExpiration(group)
+ }
+
+ private def maybeCompleteSyncExpiration(
+ group: GroupMetadata
+ ): Unit = {
+ val groupKey = GroupKey(group.groupId)
+ syncPurgatory.checkAndComplete(groupKey)
+ }
+
+ private def schedulePendingSync(
+ group: GroupMetadata
+ ): Unit = {
+ val delayedSync = new DelayedSync(this, group, group.rebalanceTimeoutMs)
Review comment:
That's a good point. I thought about it as well. Given that the goal of this PR is to protect us against misbehaving or buggy clients, I think that it is OK to allow for a full max.poll.interval.ms in between Join and Sync. Practically, I have found this approach a bit harder to test as we end up with both a DelayedJoin and a DelayedSync in parallel and they both rely on the same rebalance timeout. It is a little easier to reason about them when the are disjoint. That being said, I don't feel strong either ways.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org