kgo: rejoin a classic group when a commit returns a fatal member error

If the broker stops recognizing a member (session expired during a
network blip, or the group rebalanced without us), OffsetCommit fails
with UNKNOWN_MEMBER_ID or ILLEGAL_GENERATION. Previously we only logged
the error and kept consuming as a zombie until the heartbeat loop
noticed the dead session, up to a full heartbeat interval later.
Trigger the rejoin from the commit path immediately instead, matching
the classic Java consumer.

This is deliberately classic-only: in 848 mode a forced rejoin does not
rejoin (the session restarts with the same member id and epoch), and
with autocommit it livelocks against the session-ending sync commit.
848 fencing detection belongs to the heartbeat loop, which also matches
the Java next-gen consumer.

Closes #1326

Co-authored-by: v14dis14v <vladislav.reshetow@yandex.ru>

Author: twmb

pkg/kfake/behavior_test.go

@@ -6064,3 +6064,95 @@ func TestShareGroupRecyclePoolAliasing(t *testing.T) {
 		}
 	}
 }
+
+// injectOffsetCommitError returns a one-shot ControlKey callback that
+// fails every partition of the next OffsetCommit with the given error
+// code. Returning true pops the control after first use.
+func injectOffsetCommitError(code int16) func(kmsg.Request) (kmsg.Response, error, bool) {
+	return func(kreq kmsg.Request) (kmsg.Response, error, bool) {
+		req := kreq.(*kmsg.OffsetCommitRequest)
+		resp := req.ResponseKind().(*kmsg.OffsetCommitResponse)
+		for _, t := range req.Topics {
+			rt := kmsg.NewOffsetCommitResponseTopic()
+			rt.Topic = t.Topic
+			rt.TopicID = t.TopicID
+			for _, p := range t.Partitions {
+				rp := kmsg.NewOffsetCommitResponseTopicPartition()
+				rp.Partition = p.Partition
+				rp.ErrorCode = code
+				rt.Partitions = append(rt.Partitions, rp)
+			}
+			resp.Topics = append(resp.Topics, rt)
+		}
+		return resp, nil, true
+	}
+}
+
+// TestCommitFatalMemberErrorTriggersRejoin verifies that a classic
+// (non-848) group member rejoins immediately when an OffsetCommit
+// returns an error meaning the broker no longer recognizes the member.
+// Without the fix, the consumer would zombie along (consuming but
+// unable to commit) until the heartbeat loop noticed the dead session
+// up to a heartbeat interval later.
+func TestCommitFatalMemberErrorTriggersRejoin(t *testing.T) {
+	t.Parallel()
+	for _, test := range []struct {
+		name string
+		code int16
+	}{
+		{"UnknownMemberID", kerr.UnknownMemberID.Code},
+		{"IllegalGeneration", kerr.IllegalGeneration.Code},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			t.Parallel()
+			topic := "t-commit-fatal-" + test.name
+			group := "g-commit-fatal-" + test.name
+			const nRecords = 10
+
+			c := newCluster(t, kfake.NumBrokers(1), kfake.SeedTopics(1, topic))
+			produceNStrings(t, newPlainClient(t, c), topic, nRecords)
+
+			cl := newPlainClient(t, c,
+				kgo.ConsumeTopics(topic),
+				kgo.ConsumerGroup(group),
+				kgo.ConsumeResetOffset(kgo.NewOffset().AtStart()),
+				kgo.FetchMaxWait(250*time.Millisecond),
+				kgo.DisableAutoCommit(),
+			)
+			consumeN(t, cl, nRecords, 10*time.Second)
+
+			// Count JoinGroups to observe the rejoin; the group is
+			// stable now, so any join from here on is the rejoin.
+			var joins atomic.Int64
+			c.ControlKey(int16(kmsg.JoinGroup), func(kmsg.Request) (kmsg.Response, error, bool) {
+				c.KeepControl()
+				joins.Add(1)
+				return nil, nil, false
+			})
+
+			c.ControlKey(int16(kmsg.OffsetCommit), injectOffsetCommitError(test.code))
+
+			ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+			defer cancel()
+			cl.CommitUncommittedOffsets(ctx) // fails with the injected error; the fix reacts to it
+
+			deadline := time.Now().Add(10 * time.Second)
+			for joins.Load() == 0 && time.Now().Before(deadline) {
+				time.Sleep(50 * time.Millisecond)
+			}
+			if joins.Load() == 0 {
+				t.Fatal("timed out waiting for a rejoin after a fatal commit error")
+			}
+
+			// The member must fully recover: new records are consumable
+			// and committable after the rejoin.
+			produceNStrings(t, newPlainClient(t, c), topic, nRecords)
+			if got := consumeN(t, cl, nRecords, 15*time.Second); len(got) != nRecords {
+				t.Fatalf("expected %d records after recovery, got %d", nRecords, len(got))
+			}
+			if err := cl.CommitUncommittedOffsets(context.Background()); err != nil {
+				t.Fatalf("commit after recovery: %v", err)
+			}
+		})
+	}
+}

pkg/kgo/consumer_group.go

@@ -3159,6 +3159,7 @@ func (g *groupConsumer) commit(
 	req.Generation = generation
 	req.MemberID = memberID
 	req.InstanceID = g.cfg.instanceID
+	is848 := g.is848 // g.mu is held, per the function comment above
 
 	if ctx.Done() != nil {
 		go func() {
@@ -3401,11 +3402,67 @@ func (g *groupConsumer) commit(
 		// original request, not the wire-filtered one.
 		req.Topics = origReqTopics
 
+		// If the broker no longer recognizes our member (the session
+		// expired during a network blip, or the group rebalanced
+		// without us), every commit from here on fails identically
+		// while we keep consuming as a zombie, until the heartbeat
+		// loop sees the same error up to a full heartbeat interval
+		// later. Rejoin immediately instead; the join itself repairs
+		// the session (joinAndSync clears the member id and retries
+		// if the broker rejects the join with UnknownMemberID).
+		//
+		// Classic protocol only: in 848 mode, a forced "rejoin" does
+		// not actually rejoin. The heartbeat loop treats the signal
+		// as RebalanceInProgress and the manage loop restarts the
+		// session with the same member id and epoch; only a
+		// heartbeat error resets the member to epoch 0. Worse, with
+		// autocommit the restart livelocks: ending a session runs
+		// the default revoke, which sync-commits uncommitted
+		// offsets; that commit fails with the same fatal error and
+		// re-queues the rejoin signal; the restarted session then
+		// consumes the queued signal before its first heartbeat
+		// timer can fire, and the cycle repeats forever without a
+		// single heartbeat reaching the broker. Classic does not
+		// loop because joinAndSync drains rejoinCh before joining
+		// and the join re-registers us. For 848 we leave fatal
+		// member errors to the heartbeat loop, matching the Java
+		// clients: the classic Java consumer rejoins from the commit
+		// path, while the next-gen one leaves fencing detection to
+		// the heartbeat.
+		if !is848 {
+			if fatalErr := commitHasFatalMemberError(resp); fatalErr != nil {
+				g.cfg.logger.Log(LogLevelInfo, "offset commit returned a fatal group member error, triggering rejoin",
+					"group", g.cfg.group,
+					"err", fatalErr,
+				)
+				g.rejoin(fmt.Sprintf("offset commit error: %s", fatalErr))
+			}
+		}
+
 		g.updateCommitted(req, resp)
 		onDone(g.cl, req, resp, nil)
 	}()
 }
 
+// commitHasFatalMemberError returns the first per-partition error that
+// means the broker no longer recognizes this member's session. The
+// broker validates membership before any per-partition handling, so
+// these errors arrive on every partition or none. FencedInstanceID is
+// deliberately not included: it means another instance with our
+// instance id has taken over, and rejoining would fight that instance
+// for the group slot rather than repair anything.
+func commitHasFatalMemberError(resp *kmsg.OffsetCommitResponse) error {
+	for i := range resp.Topics {
+		for _, p := range resp.Topics[i].Partitions {
+			switch p.ErrorCode {
+			case kerr.UnknownMemberID.Code, kerr.IllegalGeneration.Code:
+				return kerr.ErrorForCode(p.ErrorCode)
+			}
+		}
+	}
+	return nil
+}
+
 type reNews struct {
 	added   map[string][]string
 	skipped []string