You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: groups.go
    path of file: ./repos/franz-go/pkg/kfake
    
    the code of the file until where you have to start completion: package kfake

import (
	"bytes"
	"fmt"
	"sync"
	"time"

	"github.com/twmb/franz-go/pkg/kerr"
	"github.com/twmb/franz-go/pkg/kmsg"
)

// TODO instance IDs
// TODO persisting groups so commits can happen to client-managed groups
//      we need lastCommit, and need to better prune empty groups

type (
	groups struct {
		c  *Cluster
		gs map[string]*group
	}

	group struct {
		c    *Cluster
		gs   *groups
		name string

		state groupState

		leader  string
		members map[string]*groupMember
		pending map[string]*groupMember

		commits tps[offsetCommit]

		generation   int32
		protocolType string
		protocols    map[string]int
		protocol     string

		reqCh     chan *clientReq
		controlCh chan func()

		nJoining int

		tRebalance *time.Timer

		quit   sync.Once
		quitCh chan struct{}
	}

	groupMember struct {
		memberID   string
		clientID   string
		clientHost string

		join *kmsg.JoinGroupRequest // the latest join request

		// waitingReply is non-nil if a client is waiting for a reply
		// from us for a JoinGroupRequest or a SyncGroupRequest.
		waitingReply *clientReq

		assignment []byte

		t    *time.Timer
		last time.Time
	}

	offsetCommit struct {
		offset      int64
		leaderEpoch int32
		metadata    *string
	}

	groupState int8
)

const (
	groupEmpty groupState = iota
	groupStable
	groupPreparingRebalance
	groupCompletingRebalance
	groupDead
)

func (gs groupState) String() string {
	switch gs {
	case groupEmpty:
		return "Empty"
	case groupStable:
		return "Stable"
	case groupPreparingRebalance:
		return "PreparingRebalance"
	case groupCompletingRebalance:
		return "CompletingRebalance"
	case groupDead:
		return "Dead"
	default:
		return "Unknown"
	}
}

func (c *Cluster) coordinator(id string) *broker {
	gen := c.coordinatorGen.Load()
	n := hashString(fmt.Sprint("%d", gen)+"\x00\x00"+id) % uint64(len(c.bs))
	return c.bs[n]
}

func (c *Cluster) validateGroup(creq *clientReq, group string) *kerr.Error {
	switch key := kmsg.Key(creq.kreq.Key()); key {
	case kmsg.Offset
    
func (gs *groups) handleOffsetFetch(creq *clientReq) *kmsg.OffsetFetchResponse {
	req := creq.kreq.(*kmsg.OffsetFetchRequest)
	resp := req.ResponseKind().(*kmsg.OffsetFetchResponse)

	if req.Version <= 7 {
		rg := kmsg.NewOffsetFetchRequestGroup()
		rg.Group = req.Group
		if req.Topics != nil {
			rg.Topics = make([]kmsg.OffsetFetchRequestGroupTopic, len(req.Topics))
		}
		for _, t := range req.Topics {
			rt := kmsg.NewOffsetFetchRequestGroupTopic()
			rt.Topic = t.Topic
			rt.Partitions = t.Partitions
			rg.Topics = append(rg.Topics, rt)
		}
		req.Groups = append(req.Groups, rg)

		defer func() {
			g0 := resp.Groups[0]
			resp.ErrorCode = g0.ErrorCode
			for _, t := range g0.Topics {
				st := kmsg.NewOffsetFetchResponseTopic()
				st.Topic = t.Topic
				for _, p := range t.Partitions {
					sp := kmsg.NewOffsetFetchResponseTopicPartition()
					sp.Partition = p.Partition
					sp.Offset = p.Offset
					sp.LeaderEpoch = p.LeaderEpoch
					sp.Metadata = p.Metadata
					sp.ErrorCode = p.ErrorCode
					st.Partitions = append(st.Partitions, sp)
				}
				resp.Topics = append(resp.Topics, st)
			}
		}()
	}

	doneg := func(name string) *kmsg.OffsetFetchResponseGroup {
		sg := kmsg.NewOffsetFetchResponseGroup()
		sg.Group = name
		resp.Groups = append(resp.Groups, sg)
		return &resp.Groups[len(resp.Groups)-1]
	}

	for _, rg := range req.Groups {
		sg := doneg(rg.Group)
		if kerr := gs.c.validateGroup(creq, rg.Group); kerr != nil {
			sg.ErrorCode = kerr.Code
			continue
		}
		g, ok := gs.gs[rg.Group]
		if !ok {
			sg.ErrorCode = kerr.GroupIDNotFound.Code
			continue
		}
		if !g.waitControl(func() {
			if rg.Topics == nil {
				for t, ps := range g.commits {
					st := kmsg.NewOffsetFetchResponseGroupTopic()
					st.Topic = t
					for p, c := range ps {
						sp := kmsg.NewOffsetFetchResponseGroupTopicPartition()
						sp.Partition = p
						sp.Offset = c.offset
						sp.LeaderEpoch = c.leaderEpoch
						sp.Metadata = c.metadata
						st.Partitions = append(st.Partitions, sp)
					}
					sg.Topics = append(sg.Topics, st)
				}
			} else {
				for _, t := range rg.Topics {
					st := kmsg.NewOffsetFetchResponseGroupTopic()
					st.Topic = t.Topic
					for _, p := range t.Partitions {
						sp := kmsg.NewOffsetFetchResponseGroupTopicPartition()
						sp.Partition = p
						c, ok := g.commits.getp(t.Topic, p)
						if !ok {
							sp.Offset = -1
							sp.LeaderEpoch = -1
						} else {
							sp.Offset = c.offset
							sp.LeaderEpoch = c.leaderEpoch
							sp.Metadata = c.metadata
						}
						st.Partitions = append(st.Partitions, sp)
					}
					sg.Topics = append(sg.Topics, st)
				}
			}
		}) {
			sg.ErrorCode = kerr.GroupIDNotFound.Code
		}
	}
	return resp
}