KAFKA-8220 & KIP-345 part-3: Avoid kicking out members through rebalance timeout #6666

abbccdda · 2019-05-02T19:01:16Z

To make consumer group members more persist, we want to avoid kick-out unjoined members through rebalance timeout. The only exception is when leader fails to join, because we will at risk of no assignment computed during sync stage. The choice will be kicking off non-responsive leader and choose a new leader if possible.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

abbccdda · 2019-05-03T17:16:28Z

Retest this please

abbccdda · 2019-05-05T06:33:55Z

@guozhangwang @hachikuji when you got time :)

rajinisivaram

@abbccdda Thanks for the PR. Changes look good to me, but it will be good to get a review from @hachikuji before merging in case I missed something.

rajinisivaram

@abbccdda Thanks for the PR. Changes look good to me, but it will be good to get a review from @hachikuji before merging in case I missed something.

abbccdda · 2019-05-07T22:22:58Z

@rajinisivaram Thanks a lot for the review! Yea, pinging @hachikuji @guozhangwang again since we got your green light lol.

hachikuji

Thanks @abbccdda. I just had a question.

core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala

hachikuji · 2019-05-08T22:33:08Z

core/src/main/scala/kafka/coordinator/group/GroupMetadata.scala

@@ -244,6 +244,25 @@ private[group] class GroupMetadata(val groupId: String, initialState: GroupState
      leaderId = members.keys.headOption
  }

+  def maybeElectNewLeader() {
+    leaderId match {


You can use foreach instead of match if there is nothing to do in the None case

core/src/main/scala/kafka/coordinator/group/GroupMetadata.scala

abbccdda · 2019-05-09T23:55:17Z

Retest this please

hachikuji

Thanks, left a few more comments.

core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala

hachikuji · 2019-05-14T00:05:45Z

core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala

+        // until session timeout removes all the non-responsive members.
+        joinPurgatory.tryCompleteElseWatch(
+          new DelayedJoin(this, group, group.rebalanceTimeoutMs),
+          Seq(group.allMembers.headOption)


Not sure I get this. Why are we using the memberId as the purgatory key?

I think we could use a dummy key here.

In fact, I do realize the purpose here. If this is last delayed join expiration, we will not trigger onJoinComplete again only depending on heartbeat timeout. So we have to constantly check for group emptiness

core/src/main/scala/kafka/coordinator/group/GroupMetadata.scala

abbccdda · 2019-05-14T18:44:41Z

core/src/test/scala/unit/kafka/coordinator/group/GroupCoordinatorTest.scala

+    assertGroupState(groupState = PreparingRebalance)
+
+    timer.advanceClock(DefaultRebalanceTimeout + 1)
+    // Only static leader is maintained, and group is stuck at PreparingRebalance stage


@hachikuji This test case covers the case where all static members are not joined but haven't session timeout yet. No new generation shall be bumped.

hachikuji

Thanks for the updates. Just a few more comments.

core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala

core/src/main/scala/kafka/coordinator/group/GroupMetadata.scala

abbccdda · 2019-05-15T15:10:27Z

Retest this please

hachikuji

Thanks, just a couple more comments.

core/src/main/scala/kafka/coordinator/group/GroupMetadata.scala

core/src/test/scala/unit/kafka/coordinator/group/GroupCoordinatorTest.scala

hachikuji · 2019-05-15T23:55:11Z

core/src/test/scala/unit/kafka/coordinator/group/GroupCoordinatorTest.scala

+    assertEquals(2, getGroup(groupId).generationId)
+    assertGroupState(groupState = PreparingRebalance)
+
+    timer.advanceClock(DefaultRebalanceTimeout + 1)


I guess another case is if the static member rejoins instead of timing out. Do you think this is worth another test case?

I think we have many static member rejoining during PrepareRebalance tests. This should be ok to skip.

Yeah, my thought is to exercise the "dummy" rebalance logic completely.

Sounds good, I got a new test case branching from the existing one @hachikuji

hachikuji

LGTM. Thanks for the patch.

guozhangwang

LGTM! Just some minor questions.

guozhangwang · 2019-05-17T18:09:55Z

core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala

+        // of rebalance preparing stage, and send out another delayed operation
+        // until session timeout removes all the non-responsive members.
+        error(s"Group ${group.groupId} could not complete rebalance because no members rejoined")
+        joinPurgatory.tryCompleteElseWatch(


Personally I'd prefer to have such logic in DelayedJoin instead, i.e. to re-write the current

override def onComplete() = coordinator.onCompleteJoin(group)

code. Similar to the extended InitialDelayedJoin#onComplete.

guozhangwang · 2019-05-17T18:16:24Z

core/src/test/scala/unit/kafka/coordinator/group/GroupCoordinatorTest.scala

@@ -480,11 +480,11 @@ class GroupCoordinatorTest extends JUnitSuite {

  @Test
  def staticMemberRejoinWithLeaderIdAndKnownMemberId() {
-    val rebalanceResult = staticMembersJoinAndRebalance(leaderInstanceId, followerInstanceId)
+    val rebalanceResult = staticMembersJoinAndRebalance(leaderInstanceId, followerInstanceId, sessionTimeout = DefaultRebalanceTimeout / 2)


Not clear why we want to override to half of the value?

We are just trying to make sure we timeout the follower instance because default session timeout = default rebalance timeout

guozhangwang · 2019-05-17T18:16:51Z

core/src/test/scala/unit/kafka/coordinator/group/GroupCoordinatorTest.scala

@@ -522,19 +522,21 @@ class GroupCoordinatorTest extends JUnitSuite {

  @Test
  def staticMemberRejoinWithFollowerIdAndChangeOfProtocol() {
-    val rebalanceResult = staticMembersJoinAndRebalance(leaderInstanceId, followerInstanceId)
+    val rebalanceResult = staticMembersJoinAndRebalance(leaderInstanceId, followerInstanceId, sessionTimeout = DefaultSessionTimeout * 2)


Ditto, why double the value here?

Doubling is aiming to reproduce the scenario where rebalanceTImeout < sessionTimeout. I could write comments for both scenarios.

guozhangwang · 2019-05-17T18:18:05Z

core/src/test/scala/unit/kafka/coordinator/group/GroupCoordinatorTest.scala

+  }
+
+  @Test
+  def testStaticMemberFollowerFailToRejoinBeforeRebalanceTimeout() {


Thanks for adding these two test cases for improving the coverage!

guozhangwang · 2019-05-17T18:18:50Z

core/src/test/scala/unit/kafka/coordinator/group/GroupCoordinatorTest.scala

-    EasyMock.reset(replicaManager)
-    heartbeatResult = heartbeat(groupId, firstMemberId, firstGenerationId)
-    assertEquals(Errors.REBALANCE_IN_PROGRESS, heartbeatResult)
+    var expectedResultList = List(Errors.REBALANCE_IN_PROGRESS, Errors.REBALANCE_IN_PROGRESS)


Nice refactoring.

…timeout (apache#6666) To make static consumer group members more persistent, we want to avoid kicking out unjoined members through rebalance timeout. Essentially we allow static members to participate in a rebalance using their old subscription without sending a JoinGroup. The only catch is that an unjoined static member might be the current group leader, and we may need to elect a different leader. Reviewers: Rajini Sivaram <rajinisivaram@googlemail.com>, Jason Gustafson <jason@confluent.io>

abbccdda added 2 commits May 2, 2019 11:56

avoid rebalance timeout kicking off

e6f9523

fix unit test

18e3cd4

rajinisivaram reviewed May 7, 2019

View reviewed changes

hachikuji reviewed May 8, 2019

View reviewed changes

core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala Outdated Show resolved Hide resolved

core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala Outdated Show resolved Hide resolved

new way

43332d0

hachikuji reviewed May 8, 2019

View reviewed changes

core/src/main/scala/kafka/coordinator/group/GroupMetadata.scala Outdated Show resolved Hide resolved

abbccdda added 2 commits May 9, 2019 12:58

delay rebalance

b57e7cb

more fix

e5e2b9a

abbccdda changed the title ~~KAFKA-8220: Avoid kicking out members through rebalance timeout~~ KAFKA-8220 & KIP-345 part-3: Avoid kicking out members through rebalance timeout May 10, 2019

hachikuji reviewed May 14, 2019

View reviewed changes

add new test

11d8866

abbccdda commented May 14, 2019

View reviewed changes

abbccdda force-pushed the rebalance_timeout branch from ea8cfcc to 11d8866 Compare May 14, 2019 18:44

hachikuji reviewed May 15, 2019

View reviewed changes

address Jason's comments

bc9832d

abbccdda force-pushed the rebalance_timeout branch from d1f4bca to bc9832d Compare May 15, 2019 04:11

hachikuji reviewed May 15, 2019

View reviewed changes

address more comments

16de0b4

abbccdda force-pushed the rebalance_timeout branch from 82bd222 to 16de0b4 Compare May 16, 2019 01:44

hachikuji approved these changes May 16, 2019

View reviewed changes

hachikuji merged commit 6e6dcce into apache:trunk May 16, 2019

guozhangwang reviewed May 17, 2019

View reviewed changes

Nevon mentioned this pull request Sep 22, 2020

Group Instance ID support tulios/kafkajs#884

Open

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-8220 & KIP-345 part-3: Avoid kicking out members through rebalance timeout #6666

KAFKA-8220 & KIP-345 part-3: Avoid kicking out members through rebalance timeout #6666

abbccdda commented May 2, 2019

abbccdda commented May 3, 2019

abbccdda commented May 5, 2019

rajinisivaram left a comment

rajinisivaram left a comment

abbccdda commented May 7, 2019

hachikuji left a comment

hachikuji May 8, 2019

abbccdda commented May 9, 2019

hachikuji left a comment

hachikuji May 14, 2019

abbccdda May 14, 2019

abbccdda May 14, 2019

abbccdda May 14, 2019 •

edited

hachikuji left a comment

abbccdda commented May 15, 2019

hachikuji left a comment

hachikuji May 15, 2019

abbccdda May 16, 2019

hachikuji May 16, 2019

abbccdda May 16, 2019

hachikuji left a comment

guozhangwang left a comment

guozhangwang May 17, 2019

guozhangwang May 17, 2019

abbccdda May 17, 2019

guozhangwang May 17, 2019

abbccdda May 17, 2019

guozhangwang May 17, 2019

guozhangwang May 17, 2019

KAFKA-8220 & KIP-345 part-3: Avoid kicking out members through rebalance timeout #6666

KAFKA-8220 & KIP-345 part-3: Avoid kicking out members through rebalance timeout #6666

Conversation

abbccdda commented May 2, 2019

Committer Checklist (excluded from commit message)

abbccdda commented May 3, 2019

abbccdda commented May 5, 2019

rajinisivaram left a comment

Choose a reason for hiding this comment

rajinisivaram left a comment

Choose a reason for hiding this comment

abbccdda commented May 7, 2019

hachikuji left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

abbccdda commented May 9, 2019

hachikuji left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

abbccdda May 14, 2019 • edited

Choose a reason for hiding this comment

hachikuji left a comment

Choose a reason for hiding this comment

abbccdda commented May 15, 2019

hachikuji left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hachikuji left a comment

Choose a reason for hiding this comment

guozhangwang left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

abbccdda May 14, 2019 •

edited