KAFKA-9232: Coordinator new member heartbeat completion does not work for JoinGroup v3 by ableegoldman · Pull Request #7753 · apache/kafka

ableegoldman · 2019-11-27T02:46:17Z

The v3 JoinGroup logic does not properly complete the initial heartbeat for new members, which then expires after the static 5 minute timeout. The core issue is in the shouldKeepAlive method, which returns false when it should return true.

ableegoldman · 2019-11-27T02:46:26Z

@hachikuji

ableegoldman · 2019-11-27T02:47:34Z

  def shouldKeepAlive(deadlineMs: Long): Boolean = {
    if (isAwaitingJoin)
      !isNew || latestHeartbeat + GroupCoordinator.NewMemberJoinTimeoutMs > deadlineMs
-    else awaitingSyncCallback != null ||


replaced awaitingSyncCallback != null with isAwaitingSync

ableegoldman · 2019-11-27T02:52:53Z

    if (isAwaitingJoin)
      !isNew || latestHeartbeat + GroupCoordinator.NewMemberJoinTimeoutMs > deadlineMs
-    else awaitingSyncCallback != null ||
+    else if (isNew || isAwaitingSync)


In onCompleteJoin, we call

group.maybeInvokeJoinCallback(member, joinResult)

completeAndScheduleNextHeartbeatExpiration(group, member)

member.isNew = false

In 1. we ultimately set member.isAwaitingJoin to false, so we never reach the first branch in this case.

I refactored this a bit to reflect what (I think) makes sense: we want to always return true in the special cases of a) awaiting Sync, and b) between awaiting the Join and awaiting the Sync, ie, when isNew = true -- otherwise, we fall back to the usual case, ie, just return whether we've seen a heartbeat within the deadline or not

Alternatively we could leave this method alone and change the ordering in onCompleteJoin to

member.isNew = false
completeAndScheduleNextHeartbeatExpiration(group, member)
group.maybeInvokeJoinCallback(member, joinResult)

I think this also makes sense, and might be a bit more clear, but I'd want to verify that it's ok to move the heartbeat completion to before the group.maybeInvokeJoinCallback

I'm for simplifying if possible. What I am thinking is the following:

if (isNew) // New members are expired after the static join timeout latestHeartbeat + GroupCoordinator.NewMemberJoinTimeoutMs > deadlineMs else if (isAwaitingJoin || isAwaitingSync) // Don't remove members as long as they have a request in purgatory true else // Otherwise check for session expiration latestHeartbeat + sessionTimeoutMs > deadlineMs

Then I think this works with the reordering you suggested above:

member.isNew = false completeAndScheduleNextHeartbeatExpiration(group, member) group.maybeInvokeJoinCallback(member, joinResult)

Upon rebalance completion, the first check will fail since we set isNew to false and the second check will succeed because we haven't cleared the join callback yet. What do you think?

Ok yeah I like that, the shouldKeepAlive logic is much easier to follow

ijuma · 2019-11-27T03:56:16Z

@ableegoldman do you know when we regressed?

hachikuji · 2019-11-27T17:52:49Z

@ijuma I think it was introduced in #5962 which was in 2.1. The bug doesn't affect newer versions of the protocol.

ableegoldman · 2019-11-27T20:17:54Z

@hachikuji updated the PR, let me know if we want to test this beyond just the usual unit/integration tests or if it's good as is

hachikuji · 2019-11-27T22:41:00Z

@ableegoldman I'd be happy with unit tests.

This reverts commit 74a9fe9.

hachikuji · 2019-12-19T19:43:32Z

-      !isNew || latestHeartbeat + GroupCoordinator.NewMemberJoinTimeoutMs > deadlineMs
-    else awaitingSyncCallback != null ||
+    if (isNew)
+    // New members are expired after the static join timeout


nit: can we fix the indentation? i think braces are fine

hachikuji

Thanks, looks good. Left some small suggestions on the test case.

hachikuji · 2019-12-20T02:20:06Z

+  @Test
+  def testCompleteNewMemberTimeoutHeartbeat(): Unit = {
+    // New member joins the group
+    var responseFuture = sendJoinGroup(groupId, JoinGroupRequest.UNKNOWN_MEMBER_ID, protocolType, protocols, None, DefaultSessionTimeout, DefaultRebalanceTimeout, false)


nit: could be val

hachikuji · 2019-12-20T02:31:52Z

+    val joinResult = Await.result(responseFuture, Duration(DefaultRebalanceTimeout + 100, TimeUnit.MILLISECONDS))
+    val group = groupCoordinator.groupManager.getGroup(groupId).get
+    assertEquals(Errors.NONE, joinResult.error)
+    assertEquals(0, group.allMemberMetadata.count(_.isNew))


Might be worth asserting latestHeartbeat as well?

hachikuji · 2019-12-20T02:34:25Z

+    assertEquals(0, group.allMemberMetadata.count(_.isNew))
+
+    // Make sure the delayed heartbeat based on the new member timeout has been completed
+    assertEquals(1, groupCoordinator.heartbeatPurgatory.numDelayed)


If we wanted to go a tad further, we could advance the clock by the session timeout and assert the member gets kicked out. That ensures that the heartbeat left behind has the right timeout.

hachikuji

LGTM. Thanks for the fix!

hachikuji · 2019-12-21T19:52:39Z

retest this please

hachikuji · 2019-12-23T16:28:49Z

retest this please

hachikuji · 2019-12-23T19:08:14Z

retest this please

…p v3 and below (#7753) The v3 JoinGroup logic does not properly complete the initial heartbeat for new members, which then expires after the static 5 minute timeout if the member does not rejoin. The core issue is in the `shouldKeepAlive` method, which returns false when it should return true because of an inconsistent timeout check. Reviewers: Jason Gustafson <jason@confluent.io>

…p v3 and below (apache#7753) The v3 JoinGroup logic does not properly complete the initial heartbeat for new members, which then expires after the static 5 minute timeout if the member does not rejoin. The core issue is in the `shouldKeepAlive` method, which returns false when it should return true because of an inconsistent timeout check. Reviewers: Jason Gustafson <jason@confluent.io>

return true when isNew

61a2acf

ableegoldman commented Nov 27, 2019

View reviewed changes

reorder calls in onCompleteJoin

74a9fe9

ableegoldman added 3 commits December 11, 2019 15:16

Revert "reorder calls in onCompleteJoin"

8392273

This reverts commit 74a9fe9.

typo

151477f

should fall to new member timeout

321fb1c

hachikuji reviewed Dec 19, 2019

View reviewed changes

ableegoldman added 2 commits December 19, 2019 16:47

add unit test

18d441d

indentation

d4d9209

hachikuji reviewed Dec 20, 2019

View reviewed changes

refactor test

69477f8

hachikuji approved these changes Dec 20, 2019

View reviewed changes

hachikuji merged commit 6b6b36c into apache:trunk Dec 23, 2019

ableegoldman deleted the 9232-coordinator-heartbeat-bug branch June 26, 2020 22:38

Conversation

ableegoldman commented Nov 27, 2019

Uh oh!

ableegoldman commented Nov 27, 2019

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ijuma commented Nov 27, 2019

Uh oh!

hachikuji commented Nov 27, 2019

Uh oh!

ableegoldman commented Nov 27, 2019

Uh oh!

hachikuji commented Nov 27, 2019

Uh oh!

Choose a reason for hiding this comment

Uh oh!

hachikuji left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

hachikuji left a comment

Choose a reason for hiding this comment

Uh oh!

hachikuji commented Dec 21, 2019

Uh oh!

hachikuji commented Dec 23, 2019

Uh oh!

hachikuji commented Dec 23, 2019

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants