fetches.EachDistinctPartition about franz-go HOT 16 CLOSED

fetches.EachPartition(func(p FetchTopicPartition) {
   doSomethingWithDistinctPartition(p.Records)
})

// FirstPartitionRecords returns the first record from all partitions in all fetches,
// and returns what remains in the fetches.
func FirstPartitionRecords(fs kgo.Fetches) (kgo.Fetches, []*kgo.Record) {
  // todo
}

type PollFuncChoice uint8

const (
  // PollFuncKeep keeps a record, returning it in fetches when PollFunc returns.
  PollFuncKeep PollFuncChoice = iota

  // PollFuncDiscard discards a record, advancing the client past it.
  PollFuncDiscard

  // PollFuncDiscardPartition discards the rest of the records in the current partition.
  PollFuncDiscardPartition

  // PollFuncDiscardTopic discards the rest of the records in the current topic.
  PollFuncDiscardTopic

  // PollFuncSkipPartition skips processing the current partition; all un-processed records
  // in this partition are candidates for the next poll.
  PollFuncSkipPartition

  // PollFuncSkipTopic skips processing the current topic; all un-processed records are candidates
  // for the next poll.
  PollFuncSkipTopic

  // PollFuncKeepThenSkipPartition keeps the current record, then skips processing the rest
  // of the current partition. All unprocessed records in this partition after this kept record are
  // candidates for the next poll.
  PollFuncKeepThenSkipPartition

  // PollFuncKeepThenSkipTopic keeps the current record, then skips processing the rest
  // of the current topic. All unprocessed records in this topic after this kept record are
  // candidates for the next poll.
  PollFuncKeepThenSkipTopic
)

// PollFunc waits for records to be available, and then uses fn to determine which
// records to return, which to discard, and which to skip. If the context quits,
// this function quits. If the context if nil or already canceled, this function
// will return immediately after evaluating any currently buffered records.
//
// It is important to check all partition errors,... etc. doc copied from PollFetches.
func (cl *Client) PollFunc(ctx context.Context, fn func(*Record) PollFuncChoice) Fetches {
  // todo
}

// FlattenRecords flattens all records in all fetches into one slice.
func (fs Fetches) FlattenRecords() []*Record {
 // todo
}

diff --git a/pkg/kgo/consumer.go b/pkg/kgo/consumer.go
index f106cfe..45d744b 100644
--- a/pkg/kgo/consumer.go
+++ b/pkg/kgo/consumer.go
@@ -376,6 +376,117 @@ func (cl *Client) PollRecords(ctx context.Context, maxPollRecords int) Fetches {
 	return fetches
 }
 
+// PollFuncResult indicates how the client should process handling the
+// current record passed to your poll func.
+type PollFuncResult uint8
+
+const (
+	// PollFuncKeep keeps the current record.
+	PollFuncKeep PollFuncResult = iota
+
+	// PollFuncKeepPartition keeps the current record and the rest of the
+	// buffered records in the current record's partition.
+	PollFuncKeepPartition
+
+	// PollFuncKeepTopic keeps the current record and the rest of the
+	// buffered records in the current record's topic.
+	PollFuncKeepTopic
+
+	// PollFuncKeepRemaining keeps the current record, the rest of the
+	// buffered records, immediately returns from PollFunc.
+	PollFuncKeepRemaining
+
+	// PollFuncDiscard discards the current record.
+	PollFuncDiscard
+
+	// PollFuncDiscardPartition discards the current record and the rest of
+	// the buffered records in the current record's partition.
+	PollFuncDiscardPartition
+
+	// PollFuncDiscardTopic discards the current record and the rest of the
+	// buffered records in the current record's topic.
+	PollFuncDiscardTopic
+
+	// PollFuncDiscardRemaining discards the current record and the rest of
+	// the buffered records and immediately returns from PollFunc.
+	PollFuncDiscardRemaining
+
+	// PollFuncSkipPartition skips the current record and the rest of the
+	// buffered records in the current record's partition. These records
+	// will be available the next time any poll function is called.
+	//
+	// There is no PollFuncSkip because, for offset consistency, if a
+	// single record is skipped, the rest of the partition must be skipped.
+	PollFuncSkipPartition
+
+	// PollFuncSkipTopic skips the current record and rest of the buffered
+	// records in the current record's topic. These records will be
+	// available the next time any poll function is called.
+	PollFuncSkipTopic
+
+	// PollFuncSkipRemaining skips the current record, the rest of the
+	// buffered records, and immediately returns from PollFunc. All skipped
+	// records will be available the next time any poll function is called.
+	PollFuncSkipRemaining
+
+	// PollFuncKeepThenDiscardPartition keeps the current record and then
+	// discards the rest of the buffered records for the current record's
+	// partition.
+	PollFuncKeepThenDiscardPartition
+
+	// PollFuncKeepThenDiscardPartition keeps the current record and then
+	// discards the rest of the buffered records for the current record's
+	// topic.
+	PollFuncKeepThenDiscardTopic
+
+	// PollFuncKeepThenDiscardPartition keeps the current record, discards
+	// the rest of the buffered records, and immediately returns from
+	// PollFunc.
+	PollFuncKeepThenDiscardRemaining
+
+	// PollFuncKeepThenSkipPartition keeps the current record and then
+	// skips the rest of the buffered records for the current record's
+	// partition. These records will be available the next time any poll
+	// function is called.
+	PollFuncKeepThenSkipPartition
+
+	// PollFuncKeepThenSkipTopic keeps the current record and then skips
+	// the rest of the buffered records for the current record's topic.
+	// These records will be available the next time any poll function is
+	// called.
+	PollFuncKeepThenSkipTopic
+
+	// PollFuncKeepThenSkipRemaining keeps the current record, skips the
+	// rest of the buffered records, and immediately returns from PollFunc.
+	// All skipped records will be available the next time any poll
+	// function is called.
+	PollFuncKeepThenSkipRemaining
+)
+
+// PollFunc is an advanced method for polling records. It is similar to
+// PollFetches or PollRecords, but allows for custom polling to determine which
+// records are returned (of all buffered records), as well as determining how
+// many records to return, and whether some records should be discarded
+// entirely without being returned.
+//
+// This function waits for fetches to be available, and then calls fn for every
+// record in all currently buffered fetches. Once each record is visited (or
+// skipped or discarded), this function returns. If the context quits before
+// any fetch is available, this function quits. If the context is nil or
+// already canceled, this function will run immediately against any currently
+// buffered fetches and will not wait if there are no buffered fetches.
+//
+// It is important to check all partition errors in the returned fetches. If
+// any partition has a fatal error and actually had no records, fake fetch will
+// be injected with the error.
+//
+// If the client is closing or has closed, a fake fetch will be injected that
+// has no topic, a partition of 0, and a partition error of ErrClientClosed.
+// This can be used to detect if the client is closing and to break out of a
+// poll loop.
+func (cl *Client) PollFunc(ctx context.Context, fn func(*Record) PollFuncResult) Fetches {
+}
+