kgo: add Fetches.RecordsAll to return a Go native iterator

Author: narqo

pkg/kgo/group_test.go

@@ -202,8 +202,7 @@ func (c *testConsumer) etl(etlsBeforeQuit int) {
 		}
 		netls++
 
-		for iter := fetches.RecordIter(); !iter.Done(); {
-			r := iter.Next()
+		for r := range fetches.RecordsAll() {
 			keyNum, err := strconv.Atoi(string(r.Key))
 			if err != nil {
 				c.errCh <- err

pkg/kgo/record_and_fetch.go

@@ -3,6 +3,7 @@ package kgo
 import (
 	"context"
 	"errors"
+	"iter"
 	"reflect"
 	"time"
 	"unsafe"
@@ -521,6 +522,19 @@ beforePartition:
 	}
 }
 
+// RecordsAll returns a Go native iterator that yields the records in a fetch.
+//
+// Similarly to RecordIter(), the errors should be inspected separately.
+func (fs Fetches) RecordsAll() iter.Seq[*Record] {
+	return func(yield func(*Record) bool) {
+		for iter := fs.RecordIter(); !iter.Done(); {
+			if !yield(iter.Next()) {
+				return
+			}
+		}
+	}
+}
+
 // EachPartition calls fn for each partition in Fetches.
 //
 // Partitions are not visited in any specific order, and a topic may be visited