func (fs Fetches) EachPartition(fn func(FetchTopicPartition)) {
	for _, fetch := range fs {
		for _, topic := range fetch.Topics {
			for i := range topic.Partitions {
				fn(FetchTopicPartition{
					Topic:          topic.Topic,
					FetchPartition: topic.Partitions[i],
				})
			}
		}
	}
}
EachPartition calls fn for each partition in Fetches.

Partitions are not visited in any specific order, and a topic may be visited
multiple times if it is spread across fetches.
package kgo
import (
	"reflect"
	"time"
	"unsafe"
)
type RecordHeader struct {
	Key   string
	Value []byte
}
type RecordAttrs struct {
	attrs uint8
}
func (a RecordAttrs) TimestampType() int8 {
	if a.attrs&0b1000_0000 != 0 {
		return -1
	}
	return int8(a.attrs & 0b0000_1000)
}
func (a RecordAttrs) CompressionType() uint8 {
	return a.attrs & 0b0000_0111
}
func (a RecordAttrs) IsTransactional() bool {
	return a.attrs&0b0001_0000 != 0
}
func (a RecordAttrs) IsControl() bool {
	return a.attrs&0b0010_0000 != 0
}
type Record struct {
	Key []byte
	Value []byte
	Headers []RecordHeader
	Timestamp time.Time
	Topic string
	Partition int32
	Attrs RecordAttrs
	ProducerEpoch int16
	ProducerID int64
	LeaderEpoch int32
	Offset int64
}
func (r *Record) AppendFormat(b []byte, layout string) ([]byte, error) {
	f, err := NewRecordFormatter(layout)
	if err != nil {
		return b, err
	}
	return f.AppendRecord(b, r), nil
}
func StringRecord(value string) *Record {
	var slice []byte
	slicehdr := (*reflect.SliceHeader)(unsafe.Pointer(&slice))
	slicehdr.Data = ((*reflect.StringHeader)(unsafe.Pointer(&value))).Data
	slicehdr.Len = len(value)
	slicehdr.Cap = len(value)
	return &Record{Value: slice}
}
func KeyStringRecord(key, value string) *Record {
	r := StringRecord(value)
	keyhdr := (*reflect.SliceHeader)(unsafe.Pointer(&r.Key))
	keyhdr.Data = ((*reflect.StringHeader)(unsafe.Pointer(&key))).Data
	keyhdr.Len = len(key)
	keyhdr.Cap = len(key)
	return r
}
func SliceRecord(value []byte) *Record {
	return &Record{Value: value}
}
func KeySliceRecord(key, value []byte) *Record {
	return &Record{Key: key, Value: value}
}
type FetchPartition struct {
	Partition int32
	Err error
	HighWatermark int64
	LastStableOffset int64
	LogStartOffset int64
	Records []*Record
}
func (p *FetchPartition) EachRecord(fn func(*Record)) {
	for _, r := range p.Records {
		fn(r)
	}
}
type FetchTopic struct {
	Topic string
	Partitions []FetchPartition
}
func (t *FetchTopic) EachPartition(fn func(FetchPartition)) {
	for i := range t.Partitions {
		fn(t.Partitions[i])
	}
}
type Fetch struct {
	Topics []FetchTopic
}
type Fetches []Fetch
type FetchError struct {
	Topic     string
	Partition int32
	Err       error
}
func (fs Fetches) Errors() []FetchError {
	var errs []FetchError
	fs.EachError(func(t string, p int32, err error) {
		errs = append(errs, FetchError{t, p, err})
	})
	return errs
}
func (f Fetch) hasErrorsOrRecords() bool {
	for i := range f.Topics {
		t := &f.Topics[i]
		for j := range t.Partitions {
			p := &t.Partitions[j]
			if p.Err != nil || len(p.Records) > 0 {
				return true
			}
		}
	}
	return false
}
func (fs Fetches) IsClientClosed() bool {
	for _, f := range fs {
		if len(f.Topics) == 1 && len(f.Topics[0].Partitions) == 1 && f.Topics[0].Partitions[0].Err == ErrClientClosed {
			return true
		}
	}
	return false
}
func (fs Fetches) EachError(fn func(string, int32, error)) {
	for _, f := range fs {
		for i := range f.Topics {
			ft := &f.Topics[i]
			for j := range ft.Partitions {
				fp := &ft.Partitions[j]
				if fp.Err != nil {
					fn(ft.Topic, fp.Partition, fp.Err)
				}
			}
		}
	}
}
func (fs Fetches) RecordIter() *FetchesRecordIter {
	iter := &FetchesRecordIter{fetches: fs}
	iter.prepareNext()
	return iter
}
type FetchesRecordIter struct {
	fetches []Fetch
	ti      int  
	pi      int  
	ri      int  
}
func (i *FetchesRecordIter) Done() bool {
	return len(i.fetches) == 0
}
func (i *FetchesRecordIter) Next() *Record {
	next := i.fetches[0].Topics[i.ti].Partitions[i.pi].Records[i.ri]
	i.ri++
	i.prepareNext()
	return next
}
func (i *FetchesRecordIter) prepareNext() {
beforeFetch0:
	if len(i.fetches) == 0 {
		return
	}
	fetch0 := &i.fetches[0]
beforeTopic:
	if i.ti >= len(fetch0.Topics) {
		i.fetches = i.fetches[1:]
		i.ti = 0
		goto beforeFetch0
	}
	topic := &fetch0.Topics[i.ti]
beforePartition:
	if i.pi >= len(topic.Partitions) {
		i.ti++
		i.pi = 0
		goto beforeTopic
	}
	partition := &topic.Partitions[i.pi]
	if i.ri >= len(partition.Records) {
		i.pi++
		i.ri = 0
		goto beforePartition
	}
}