func (os OffsetResponses) EachError(fn func(o OffsetResponse)) {
	for _, ps := range os {
		for _, o := range ps {
			if o.Err != nil {
				fn(o)
			}
		}
	}
}
EachError calls fn for every offset that as a non-nil error.
package kadm
import (
	"context"
	"errors"
	"fmt"
	"sort"
	"sync"
	"github.com/twmb/franz-go/pkg/kerr"
	"github.com/twmb/franz-go/pkg/kmsg"
)
type DescribedGroupMember struct {
	MemberID   string   
	InstanceID *string  
	ClientID   string   
	ClientHost string   
	Join     kmsg.GroupMemberMetadata    
	Assigned kmsg.GroupMemberAssignment  
}
func (d *DescribedGroup) AssignedPartitions() TopicsSet {
	var s TopicsSet
	for _, m := range d.Members {
		for _, t := range m.Assigned.Topics {
			s.Add(t.Topic, t.Partitions...)
		}
	}
	return s
}
type DescribedGroup struct {
	Group string  
	Coordinator BrokerDetail            
	State       string                  
	Protocol    string                  
	Members     []DescribedGroupMember  
	Err error  
}
type DescribedGroups map[string]DescribedGroup
func (ds DescribedGroups) AssignedPartitions() TopicsSet {
	var s TopicsSet
	for _, g := range ds {
		for _, m := range g.Members {
			for _, t := range m.Assigned.Topics {
				s.Add(t.Topic, t.Partitions...)
			}
		}
	}
	return s
}
func (ds DescribedGroups) Sorted() []DescribedGroup {
	s := make([]DescribedGroup, 0, len(ds))
	for _, d := range ds {
		s = append(s, d)
	}
	sort.Slice(s, func(i, j int) bool { return s[i].Group < s[j].Group })
	return s
}
func (rs DescribedGroups) On(group string, fn func(*DescribedGroup) error) (DescribedGroup, error) {
	if len(rs) > 0 {
		r, ok := rs[group]
		if ok {
			if fn == nil {
				return r, nil
			}
			return r, fn(&r)
		}
	}
	return DescribedGroup{}, errMissing(group)
}
func (ds DescribedGroups) Names() []string {
	all := make([]string, 0, len(ds))
	for g := range ds {
		all = append(all, g)
	}
	sort.Strings(all)
	return all
}
type ListedGroup struct {
	Group string  
	State string  
}
type ListedGroups map[string]ListedGroup
func (ls ListedGroups) Sorted() []ListedGroup {
	s := make([]ListedGroup, 0, len(ls))
	for _, l := range ls {
		s = append(s, l)
	}
	sort.Slice(s, func(i, j int) bool { return s[i].Group < s[j].Group })
	return s
}
func (ls ListedGroups) Groups() []string {
	all := make([]string, 0, len(ls))
	for g := range ls {
		all = append(all, g)
	}
	sort.Strings(all)
	return all
}
func (cl *Client) ListGroups(ctx context.Context, filterStates ...string) (ListedGroups, error) {
	req := kmsg.NewPtrListGroupsRequest()
	req.StatesFilter = append(req.StatesFilter, filterStates...)
	shards := cl.cl.RequestSharded(ctx, req)
	list := make(ListedGroups)
	return list, shardErrEach(req, shards, func(kr kmsg.Response) error {
		resp := kr.(*kmsg.ListGroupsResponse)
		if err := maybeAuthErr(resp.ErrorCode); err != nil {
			return err
		}
		if err := kerr.ErrorForCode(resp.ErrorCode); err != nil {
			return err
		}
		for _, g := range resp.Groups {
			list[g.Group] = ListedGroup{
				Group: g.Group,
				State: g.GroupState,
			}
		}
		return nil
	})
}
func (cl *Client) DescribeGroups(ctx context.Context, groups ...string) (DescribedGroups, error) {
	var seList *ShardErrors
	if len(groups) == 0 {
		listed, err := cl.ListGroups(ctx)
		switch {
		case err == nil:
		case errors.As(err, &seList):
		default:
			return nil, err
		}
		groups = listed.Groups()
		if len(groups) == 0 {
			return nil, err
		}
	}
	req := kmsg.NewPtrDescribeGroupsRequest()
	req.Groups = groups
	shards := cl.cl.RequestSharded(ctx, req)
	described := make(DescribedGroups)
	err := shardErrEachBroker(req, shards, func(b BrokerDetail, kr kmsg.Response) error {
		resp := kr.(*kmsg.DescribeGroupsResponse)
		for _, rg := range resp.Groups {
			if err := maybeAuthErr(rg.ErrorCode); err != nil {
				return err
			}
			g := DescribedGroup{
				Group:       rg.Group,
				Coordinator: b,
				State:       rg.State,
				Protocol:    rg.Protocol,
				Err:         kerr.ErrorForCode(rg.ErrorCode),
			}
			for _, rm := range rg.Members {
				gm := DescribedGroupMember{
					MemberID:   rm.MemberID,
					InstanceID: rm.InstanceID,
					ClientID:   rm.ClientID,
					ClientHost: rm.ClientHost,
				}
				gm.Join.ReadFrom(rm.ProtocolMetadata)
				gm.Assigned.ReadFrom(rm.MemberAssignment)
				g.Members = append(g.Members, gm)
			}
			sort.Slice(g.Members, func(i, j int) bool {
				if g.Members[i].InstanceID != nil {
					if g.Members[j].InstanceID == nil {
						return true
					}
					return *g.Members[i].InstanceID < *g.Members[j].InstanceID
				}
				if g.Members[j].InstanceID != nil {
					return false
				}
				return g.Members[i].MemberID < g.Members[j].MemberID
			})
			described[g.Group] = g
		}
		return nil
	})
	var seDesc *ShardErrors
	switch {
	case err == nil:
		return described, seList.into()
	case errors.As(err, &seDesc):
		if seList != nil {
			seDesc.Errs = append(seList.Errs, seDesc.Errs...)
		}
		return described, seDesc.into()
	default:
		return nil, err
	}
}
type DeleteGroupResponse struct {
	Group string  
	Err   error   
}
type DeleteGroupResponses map[string]DeleteGroupResponse
func (ds DeleteGroupResponses) Sorted() []DeleteGroupResponse {
	s := make([]DeleteGroupResponse, 0, len(ds))
	for _, d := range ds {
		s = append(s, d)
	}
	sort.Slice(s, func(i, j int) bool { return s[i].Group < s[j].Group })
	return s
}
func (rs DeleteGroupResponses) On(group string, fn func(*DeleteGroupResponse) error) (DeleteGroupResponse, error) {
	if len(rs) > 0 {
		r, ok := rs[group]
		if ok {
			if fn == nil {
				return r, nil
			}
			return r, fn(&r)
		}
	}
	return DeleteGroupResponse{}, errMissing(group)
}
func (cl *Client) DeleteGroups(ctx context.Context, groups ...string) (DeleteGroupResponses, error) {
	if len(groups) == 0 {
		return nil, nil
	}
	req := kmsg.NewPtrDeleteGroupsRequest()
	req.Groups = append(req.Groups, groups...)
	shards := cl.cl.RequestSharded(ctx, req)
	rs := make(map[string]DeleteGroupResponse)
	return rs, shardErrEach(req, shards, func(kr kmsg.Response) error {
		resp := kr.(*kmsg.DeleteGroupsResponse)
		for _, g := range resp.Groups {
			rs[g.Group] = DeleteGroupResponse{
				Group: g.Group,
				Err:   kerr.ErrorForCode(g.ErrorCode),
			}
		}
		return nil
	})
}
type OffsetResponse struct {
	Offset
	Err error  
}
type OffsetResponses map[string]map[int32]OffsetResponse
func (os OffsetResponses) Keep(o Offsets) {
	os.DeleteFunc(func(r OffsetResponse) bool {
		if len(o) == 0 {
			return true  
		}
		ot := o[r.Topic]
		if ot == nil {
			return true  
		}
		_, ok := ot[r.Partition]
		return !ok  
	})
}
func (os OffsetResponses) Into() Offsets {
	var i Offsets
	os.Each(func(o OffsetResponse) {
		i.Add(o.Offset)
	})
	return i
}
func (os OffsetResponses) DeleteFunc(fn func(OffsetResponse) bool) {
	for t, ps := range os {
		for p, o := range ps {
			if fn(o) {
				delete(ps, p)
			}
		}
		if len(ps) == 0 {
			delete(os, t)
		}
	}
}