func (g *groupConsumer) joinAndSync() error {
	g.cfg.logger.Log(LogLevelInfo, "joining group", "group", g.cfg.group)
	g.leader.set(false)
start:
	select {
	case <-g.rejoinCh:  
	default:
	}
	joinReq := kmsg.NewPtrJoinGroupRequest()
	joinReq.Group = g.cfg.group
	joinReq.SessionTimeoutMillis = int32(g.cfg.sessionTimeout.Milliseconds())
	joinReq.RebalanceTimeoutMillis = int32(g.cfg.rebalanceTimeout.Milliseconds())
	joinReq.ProtocolType = g.cfg.protocol
	joinReq.MemberID = g.memberID
	joinReq.InstanceID = g.cfg.instanceID
	joinReq.Protocols = g.joinGroupProtocols()
	var (
		joinResp *kmsg.JoinGroupResponse
		err      error
		joined   = make(chan struct{})
	)
	go func() {
		defer close(joined)
		joinResp, err = joinReq.RequestWith(g.ctx, g.cl)
	}()
	select {
	case <-joined:
	case <-g.ctx.Done():
		return g.ctx.Err()  
	}
	if err != nil {
		return err
	}
	restart, protocol, plan, err := g.handleJoinResp(joinResp)
	if restart {
		goto start
	}
	if err != nil {
		g.cfg.logger.Log(LogLevelWarn, "join group failed", "group", g.cfg.group, "err", err)
		return err
	}
	syncReq := kmsg.NewPtrSyncGroupRequest()
	syncReq.Group = g.cfg.group
	syncReq.Generation = g.generation
	syncReq.MemberID = g.memberID
	syncReq.InstanceID = g.cfg.instanceID
	syncReq.ProtocolType = &g.cfg.protocol
	syncReq.Protocol = &protocol
	syncReq.GroupAssignment = plan  
	var (
		syncResp *kmsg.SyncGroupResponse
		synced   = make(chan struct{})
	)
	g.cfg.logger.Log(LogLevelInfo, "syncing", "group", g.cfg.group, "protocol_type", g.cfg.protocol, "protocol", protocol)
	go func() {
		defer close(synced)
		syncResp, err = syncReq.RequestWith(g.ctx, g.cl)
	}()
	select {
	case <-synced:
	case <-g.ctx.Done():
		return g.ctx.Err()
	}
	if err != nil {
		return err
	}
	if err = g.handleSyncResp(protocol, syncResp); err != nil {
		if err == kerr.RebalanceInProgress {
			g.cfg.logger.Log(LogLevelInfo, "sync failed with RebalanceInProgress, rejoining", "group", g.cfg.group)
			goto start
		}
		g.cfg.logger.Log(LogLevelWarn, "sync group failed", "group", g.cfg.group, "err", err)
		return err
	}
	return nil
}
Joins and then syncs, issuing the two slow requests in goroutines to allow
for group cancelation to return early.
package kgo
import (
	"bytes"
	"context"
	"fmt"
	"sort"
	"strings"
	"sync"
	"time"
	"github.com/twmb/franz-go/pkg/kerr"
	"github.com/twmb/franz-go/pkg/kmsg"
)
type groupConsumer struct {
	c   *consumer  
	cl  *Client    
	cfg *cfg
	ctx        context.Context
	cancel     func()
	manageDone chan struct{}  
	cooperative bool  
	tps *topicsPartitions
	reSeen map[string]bool  
	syncCommitMu sync.RWMutex
	rejoinCh chan string  
	heartbeatForceCh chan func(error)
	lastAssigned map[string][]int32  
	nowAssigned  map[string][]int32  
	fetching map[string]map[int32]struct{}
	leader atomicBool
	offsetsAddedToTxn bool
	mu sync.Mutex
	using map[string]int  
	uncommitted uncommitted
	memberID   string
	generation int32
	commitCancel func()
	commitDone   chan struct{}
	blockAuto bool
	dying bool  
}
func (cl *Client) LeaveGroup() {
	cl.consumer.unset()
}
func (c *consumer) initGroup() {
	ctx, cancel := context.WithCancel(c.cl.ctx)
	g := &groupConsumer{
		c:   c,
		cl:  c.cl,
		cfg: &c.cl.cfg,
		ctx:    ctx,
		cancel: cancel,
		reSeen: make(map[string]bool),
		manageDone:       make(chan struct{}),
		cooperative:      c.cl.cfg.cooperative(),
		tps:              newTopicsPartitions(),
		rejoinCh:         make(chan string, 1),
		heartbeatForceCh: make(chan func(error)),
		using:            make(map[string]int),
	}
	c.g = g
	if !g.cfg.setCommitCallback {
		g.cfg.commitCallback = g.defaultCommitCallback
	}
	if g.cfg.txnID == nil {
		if !g.cfg.setRevoked {
			g.cfg.onRevoked = g.defaultRevoke
		}
		if !g.cfg.setLost {
			g.cfg.onLost = func(context.Context, *Client, map[string][]int32) {}
		}
	} else {
		g.cfg.autocommitDisable = true
	}
	for _, logOn := range []struct {
		name string
		set  *func(context.Context, *Client, map[string][]int32)
	}{
		{"OnAssigned", &g.cfg.onAssigned},
		{"OnRevoked", &g.cfg.onRevoked},
		{"OnLost", &g.cfg.onLost},
	} {
		user := *logOn.set
		name := logOn.name
		*logOn.set = func(ctx context.Context, cl *Client, m map[string][]int32) {
			var ctxExpired bool
			select {
			case <-ctx.Done():
				ctxExpired = true
			default:
			}
			if ctxExpired {
				cl.cfg.logger.Log(LogLevelDebug, "entering "+name, "with", m, "context_expired", ctxExpired)
			} else {
				cl.cfg.logger.Log(LogLevelDebug, "entering "+name, "with", m)
			}
			if user != nil {
				dup := make(map[string][]int32)
				for k, vs := range m {
					dup[k] = append([]int32(nil), vs...)
				}
				user(ctx, cl, dup)
			}
		}
	}
	if len(g.cfg.topics) > 0 {
		topics := make([]string, 0, len(g.cfg.topics))
		for topic := range g.cfg.topics {
			topics = append(topics, topic)
		}
		g.tps.storeTopics(topics)
	}
	if !g.cfg.autocommitDisable && g.cfg.autocommitInterval > 0 {
		g.cfg.logger.Log(LogLevelInfo, "beginning autocommit loop", "group", g.cfg.group)
		go g.loopCommit()
	}
}
func (g *groupConsumer) manage() {
	defer close(g.manageDone)
	g.cfg.logger.Log(LogLevelInfo, "beginning to manage the group lifecycle", "group", g.cfg.group)
	var consecutiveErrors int
	for {
		err := g.joinAndSync()
		if err == nil {
			if err = g.setupAssignedAndHeartbeat(); err != nil {
				if err == kerr.RebalanceInProgress {
					err = nil
				}
			}
		}
		if err == nil {
			consecutiveErrors = 0
			continue
		}
		hook := func() {
			g.cfg.hooks.each(func(h Hook) {
				if h, ok := h.(HookGroupManageError); ok {
					h.OnGroupManageError(err)
				}
			})
		}
		if err == context.Canceled && g.cfg.onRevoked != nil {
			g.cfg.onRevoked(g.cl.ctx, g.cl, g.nowAssigned)
		} else {
			if g.cfg.onLost != nil {
				g.cfg.onLost(g.cl.ctx, g.cl, g.nowAssigned)
			}
			hook()
		}
		{
			g.c.mu.Lock()
			g.c.assignPartitions(nil, assignInvalidateAll, nil)
			g.mu.Lock()      
			g.c.mu.Unlock()  
			g.uncommitted = nil
			g.mu.Unlock()
			g.nowAssigned = nil
			g.lastAssigned = nil
			g.fetching = nil
			g.leader.set(false)
		}
		if err == context.Canceled {  
			return
		}
		consecutiveErrors++
		backoff := g.cfg.retryBackoff(consecutiveErrors)
		g.cfg.logger.Log(LogLevelError, "join and sync loop errored",
			"group", g.cfg.group,
			"err", err,
			"consecutive_errors", consecutiveErrors,
			"backoff", backoff,
		)
		deadline := time.Now().Add(backoff)
		g.cl.waitmeta(g.ctx, backoff, "waitmeta during join & sync error backoff")
		after := time.NewTimer(time.Until(deadline))
		select {
		case <-g.ctx.Done():
			after.Stop()
			return
		case <-after.C:
		}
	}
}
func (g *groupConsumer) leave() (wait func()) {
	g.mu.Lock()
	wasDead := g.dying
	g.dying = true
	wasManaging := len(g.using) > 0
	g.mu.Unlock()
	done := make(chan struct{})
	go func() {
		defer close(done)
		g.cancel()
		if wasManaging {
			<-g.manageDone
		}
		if wasDead {
			return
		}
		if g.cfg.instanceID == nil {
			g.cfg.logger.Log(LogLevelInfo, "leaving group",
				"group", g.cfg.group,
				"member_id", g.memberID,  
			)
			req := kmsg.NewPtrLeaveGroupRequest()
			req.Group = g.cfg.group
			req.MemberID = g.memberID
			member := kmsg.NewLeaveGroupRequestMember()
			member.MemberID = g.memberID
			req.Members = append(req.Members, member)
			req.RequestWith(g.cl.ctx, g.cl)
		}
	}()
	return func() { <-done }
}
func (g *groupConsumer) diffAssigned() (added, lost map[string][]int32) {
	if g.lastAssigned == nil {
		return g.nowAssigned, nil
	}
	added = make(map[string][]int32, len(g.nowAssigned))
	lost = make(map[string][]int32, len(g.nowAssigned))
	lasts := make(map[int32]struct{}, 100)
	for topic, lastPartitions := range g.lastAssigned {
		nowPartitions, exists := g.nowAssigned[topic]
		if !exists {
			lost[topic] = lastPartitions
			continue
		}
		for _, lastPartition := range lastPartitions {
			lasts[lastPartition] = struct{}{}
		}
		for _, nowPartition := range nowPartitions {
			if _, exists := lasts[nowPartition]; !exists {
				added[topic] = append(added[topic], nowPartition)
			} else {
				delete(lasts, nowPartition)
			}
		}
		for last := range lasts {
			lost[topic] = append(lost[topic], last)
			delete(lasts, last)  
		}
	}
	for topic, nowPartitions := range g.nowAssigned {
		if _, exists := g.lastAssigned[topic]; !exists {
			added[topic] = nowPartitions
		}
	}
	return added, lost
}
type revokeStage int8
const (
	revokeLastSession = iota
	revokeThisSession
)
func (g *groupConsumer) revoke(stage revokeStage, lost map[string][]int32, leaving bool) {
	if !g.cooperative || leaving {  
		g.c.mu.Lock()
		g.c.assignPartitions(nil, assignInvalidateAll, nil)
		g.c.mu.Unlock()
		if !g.cooperative {
			g.cfg.logger.Log(LogLevelInfo, "eager consumer revoking prior assigned partitions", "group", g.cfg.group, "revoking", g.nowAssigned)
		} else {
			g.cfg.logger.Log(LogLevelInfo, "cooperative consumer revoking prior assigned partitions because leaving group", "group", g.cfg.group, "revoking", g.nowAssigned)
		}
		if g.cfg.onRevoked != nil {
			g.cfg.onRevoked(g.cl.ctx, g.cl, g.nowAssigned)
		}
		g.nowAssigned = nil
		g.mu.Lock()
		g.uncommitted = nil
		g.mu.Unlock()
		return
	}
	switch stage {
	case revokeLastSession:
	case revokeThisSession:
	}
	if len(lost) > 0 {
		lostOffsets := make(map[string]map[int32]Offset, len(lost))
		for lostTopic, lostPartitions := range lost {
			lostPartitionOffsets := make(map[int32]Offset, len(lostPartitions))
			for _, lostPartition := range lostPartitions {
				lostPartitionOffsets[lostPartition] = Offset{}
			}
			lostOffsets[lostTopic] = lostPartitionOffsets
		}
		g.c.mu.Lock()
		g.c.assignPartitions(lostOffsets, assignInvalidateMatching, g.tps)
		g.c.mu.Unlock()
	}
	if len(lost) > 0 || stage == revokeThisSession {
		if len(lost) == 0 {
			g.cfg.logger.Log(LogLevelInfo, "cooperative consumer calling onRevoke at the end of a session even though no partitions were lost", "group", g.cfg.group)
		} else {
			g.cfg.logger.Log(LogLevelInfo, "cooperative consumer calling onRevoke", "group", g.cfg.group, "lost", lost, "stage", stage)
		}
		if g.cfg.onRevoked != nil {
			g.cfg.onRevoked(g.cl.ctx, g.cl, lost)
		}
	}
	if len(lost) == 0 {  
		return
	}
	defer g.rejoin("cooperative rejoin after revoking what we lost")  
	g.mu.Lock()
	defer g.mu.Unlock()
	if g.uncommitted == nil {
		return
	}
	for lostTopic, lostPartitions := range lost {
		uncommittedPartitions := g.uncommitted[lostTopic]
		if uncommittedPartitions == nil {
			continue
		}
		for _, lostPartition := range lostPartitions {
			delete(uncommittedPartitions, lostPartition)
		}
		if len(uncommittedPartitions) == 0 {
			delete(g.uncommitted, lostTopic)
		}
	}
	if len(g.uncommitted) == 0 {
		g.uncommitted = nil
	}
}
type assignRevokeSession struct {
	prerevokeDone chan struct{}
	assignDone    chan struct{}
	revokeDone    chan struct{}
}
func newAssignRevokeSession() *assignRevokeSession {
	return &assignRevokeSession{
		prerevokeDone: make(chan struct{}),
		assignDone:    make(chan struct{}),
		revokeDone:    make(chan struct{}),
	}
}
func (s *assignRevokeSession) prerevoke(g *groupConsumer, lost map[string][]int32) <-chan struct{} {
	go func() {
		defer close(s.prerevokeDone)
		if g.cooperative && len(lost) > 0 {
			g.revoke(revokeLastSession, lost, false)
		}
	}()
	return s.prerevokeDone
}
func (s *assignRevokeSession) assign(g *groupConsumer, newAssigned map[string][]int32) <-chan struct{} {
	go func() {
		defer close(s.assignDone)
		<-s.prerevokeDone
		if g.cfg.onAssigned != nil {
			g.cfg.onAssigned(g.cl.ctx, g.cl, newAssigned)
		}
	}()
	return s.assignDone
}
func (s *assignRevokeSession) revoke(g *groupConsumer, leaving bool) <-chan struct{} {
	go func() {
		defer close(s.revokeDone)
		<-s.assignDone
		g.revoke(revokeThisSession, nil, leaving)
	}()
	return s.revokeDone
}
func (g *groupConsumer) setupAssignedAndHeartbeat() error {
	hbErrCh := make(chan error, 1)
	fetchErrCh := make(chan error, 1)
	s := newAssignRevokeSession()
	added, lost := g.diffAssigned()
	g.cfg.logger.Log(LogLevelInfo, "new group session begun", "group", g.cfg.group, "added", added, "lost", lost)
	s.prerevoke(g, lost)  
	ctx, cancel := context.WithCancel(g.ctx)
	go func() {
		defer cancel()  
		g.cfg.logger.Log(LogLevelInfo, "beginning heartbeat loop", "group", g.cfg.group)
		hbErrCh <- g.heartbeat(fetchErrCh, s)
	}()
	fetchDone := make(chan struct{})
	defer func() { <-fetchDone }()
	if len(added) > 0 {
		go func() {
			defer close(fetchDone)
			defer close(fetchErrCh)
			g.cfg.logger.Log(LogLevelInfo, "fetching offsets for added partitions", "group", g.cfg.group, "added", added)
			fetchErrCh <- g.fetchOffsets(ctx, added, lost)
		}()
	} else {
		close(fetchDone)
		close(fetchErrCh)
	}
	s.assign(g, added)
	defer func() { <-s.assignDone }()
	return <-hbErrCh
}
func (g *groupConsumer) heartbeat(fetchErrCh <-chan error, s *assignRevokeSession) error {
	ticker := time.NewTicker(g.cfg.heartbeatInterval)
	defer ticker.Stop()
	var cooperativeFastCheck <-chan time.Time
	if g.cooperative {
		cooperativeFastCheck = time.After(500 * time.Millisecond)
	}
	var metadone, revoked <-chan struct{}
	var heartbeat, didMetadone, didRevoke bool
	var lastErr error
	ctxCh := g.ctx.Done()
	for {
		var err error
		var force func(error)
		heartbeat = false
		select {
		case <-cooperativeFastCheck:
			heartbeat = true
		case <-ticker.C:
			heartbeat = true
		case force = <-g.heartbeatForceCh:
			heartbeat = true
		case why := <-g.rejoinCh:
			g.cfg.logger.Log(LogLevelInfo, "forced rejoin quitting heartbeat loop", "why", why)
			err = kerr.RebalanceInProgress
		case err = <-fetchErrCh:
			fetchErrCh = nil
		case <-metadone:
			metadone = nil
			didMetadone = true
		case <-revoked:
			revoked = nil
			didRevoke = true
		case <-ctxCh:
			ctxCh = nil
			err = context.Canceled
		}
		if heartbeat {
			g.cfg.logger.Log(LogLevelDebug, "heartbeating", "group", g.cfg.group)
			req := kmsg.NewPtrHeartbeatRequest()
			req.Group = g.cfg.group
			req.Generation = g.generation
			req.MemberID = g.memberID
			req.InstanceID = g.cfg.instanceID
			var resp *kmsg.HeartbeatResponse
			if resp, err = req.RequestWith(g.ctx, g.cl); err == nil {
				err = kerr.ErrorForCode(resp.ErrorCode)
			}
			g.cfg.logger.Log(LogLevelDebug, "heartbeat complete", "group", g.cfg.group, "err", err)
			if force != nil {
				force(err)
			}
		}
		if didMetadone && didRevoke {
			g.cfg.logger.Log(LogLevelInfo, "heartbeat loop complete", "group", g.cfg.group, "err", lastErr)
			return lastErr
		}
		if err == nil {
			continue
		}
		if lastErr == nil {
			g.cfg.logger.Log(LogLevelInfo, "heartbeat errored", "group", g.cfg.group, "err", err)
		} else {
			g.cfg.logger.Log(LogLevelInfo, "heartbeat errored again while waiting for user revoke to finish", "group", g.cfg.group, "err", err)
		}
		if !didRevoke && revoked == nil {
			if err != kerr.RebalanceInProgress && revoked == nil {
				return err
			}
			revoked = s.revoke(g, err == context.Canceled)
		}
		if !didMetadone && metadone == nil {
			waited := make(chan struct{})
			metadone = waited
			go func() {
				g.cl.waitmeta(g.ctx, g.cfg.sessionTimeout, "waitmeta after heartbeat error")
				close(waited)
			}()
		}
		lastErr = err
	}
}
func (cl *Client) ForceRebalance() {
	if g := cl.consumer.g; g != nil {
		g.rejoin("rejoin from ForceRebalance")
	}
}
func (g *groupConsumer) rejoin(why string) {
	select {
	case g.rejoinCh <- why:
	default:
	}
}