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
}
returns the difference of g.nowAssigned and g.lastAssigned.
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 }
}