fix(consumer): support JoinGroup V4

Had previously commented this out until I'd looked into the behaviour described in the protocol doc, but it looks like we should be safe to support this. Signed-off-by: Dominic Evans <[email protected]>
IBM · Aug 3, 2023 · 01ebd8c · 01ebd8c
1 parent 50609a0
commit 01ebd8c
Showing 1 changed file with 11 additions and 10 deletions.
diff --git a/consumer_group.go b/consumer_group.go
@@ -318,8 +318,10 @@ func (c *consumerGroup) newSession(ctx context.Context, topics []string, handler
 		}
 		return c.retryNewSession(ctx, topics, handler, retries, true)
 	case ErrMemberIdRequired:
-		// from JoinGroupRequest v4, if client start with empty member id,
-		// it need to get member id from response and send another join request to join group
+		// from JoinGroupRequest v4 onwards (due to KIP-394) if the client starts
+		// with an empty member id, it needs to get the assigned id from the
+		// response and send another join request with that id to actually join the
+		// group
 		c.memberID = join.MemberId
 		return c.retryNewSession(ctx, topics, handler, retries+1 /*keep retry time*/, false)
 	case ErrFencedInstancedId:
@@ -442,14 +444,13 @@ func (c *consumerGroup) joinGroupRequest(coordinator *Broker, topics []string) (
 	if c.config.Version.IsAtLeast(V2_0_0_0) {
 		req.Version = 3
 	}
-	// XXX: protocol states "Starting from version 4, the client needs to issue a
-	// second request to join group", so not enabling this until we can
-	// investigate
-	/*
-		if c.config.Version.IsAtLeast(V2_2_0_0) {
-			req.Version = 4
-		}
-	*/
+	// from JoinGroupRequest v4 onwards (due to KIP-394) the client will actually
+	// send two JoinGroupRequests, once with the empty member id, and then again
+	// with the assigned id from the first response. This is handled via the
+	// ErrMemberIdRequired case.
+	if c.config.Version.IsAtLeast(V2_2_0_0) {
+		req.Version = 4
+	}
 	if c.config.Version.IsAtLeast(V2_3_0_0) {
 		req.Version = 5
 		req.GroupInstanceId = c.groupInstanceId