kgo: fix mark <=> epoch interaction, make epoch handling more resilient (#1203)

MarkCommit{Records,Offsets} compared the epoch/offset you were marking
against the existing head epoch/offset. If the there was no internally
existing epoch/offset, or only `dirty` was set (via being consumed)
while `head` / `committed` were both the default struct value {0,0},
then trying to mark a record with a negative epoch would be ignored.

Note that returning -1 via the broker epoch requires a broker to both
(a) SUPPORT epochs, i.e. implement all the Kafka APIs with leader epoch
support, and then (b) NOT SUPPORT epochs, i.e. return / use -1
everywhere. This has only been seen against Azure Event hubs.

Anyway, now, when initializing an EpochOffset internally (for an
uncommit {dirty,head,committed}, the epoch is explicitly initialized
with -1. Further, for added robustness, MarkCommit{Records,Offsets}
only compares against existing values -- if a value does not exist,
we auto-accept the mark.

This commit also:
* Improves EpochOffset.Less to assume all negative epochs are -1
* Simplifies the logic in CommitRecords to use EpochOffset.Less
* Adds debug lines in metadata when new partitions are added
* Adds the epoch to an existing debug log line when updating uncommitted
* Sets the LeaderEpoch to -1 for records when producing, and clarifies
  in docs that -1 should have always been the case.

Author: twmb

pkg/kgo/consumer_group.go

@@ -1878,7 +1878,8 @@ type EpochOffset struct {
 // than the other if this one's epoch is less, or the epoch's are equal and
 // this one's offset is less.
 func (e EpochOffset) Less(o EpochOffset) bool {
-	return e.Epoch < o.Epoch || e.Epoch == o.Epoch && e.Offset < o.Offset
+	ee, oe := max(e.Epoch, -1), max(o.Epoch, -1)
+	return ee < oe || ee == oe && e.Offset < o.Offset
 }
 
 type uncommitted map[string]map[int32]uncommit
@@ -1925,13 +1926,18 @@ func (g *groupConsumer) updateUncommitted(fetches Fetches) {
 					final.LeaderEpoch, // -1 if old message / unknown
 					final.Offset + 1,
 				}
-				prior := topicOffsets[partition.Partition]
+				prior, ok := topicOffsets[partition.Partition]
+				if !ok {
+					uninit := EpochOffset{-1, 0}
+					uncommit := uncommit{uninit, uninit, uninit}
+					prior, topicOffsets[partition.Partition] = uncommit, uncommit
+				}
 
 				if debug {
 					if setHead {
-						fmt.Fprintf(&b, "%d{%d=>%d r%d}, ", partition.Partition, prior.head.Offset, set.Offset, len(partition.Records))
+						fmt.Fprintf(&b, "%d{%d=>%d r%d e%d}, ", partition.Partition, prior.head.Offset, set.Offset, len(partition.Records), set.Epoch)
 					} else {
-						fmt.Fprintf(&b, "%d{%d=>%d=>%d r%d}, ", partition.Partition, prior.head.Offset, prior.dirty.Offset, set.Offset, len(partition.Records))
+						fmt.Fprintf(&b, "%d{%d=>%d=>%d r%d e%d}, ", partition.Partition, prior.head.Offset, prior.dirty.Offset, set.Offset, len(partition.Records), set.Epoch)
 					}
 				}
 
@@ -2400,15 +2406,18 @@ func (cl *Client) CommitRecords(ctx context.Context, rs ...*Record) error {
 			offsets[r.Topic] = toffsets
 		}
 
+		set := EpochOffset{
+			r.LeaderEpoch,
+			r.Offset + 1, // need to advice to next offset to move forward
+		}
+
 		if at, exists := toffsets[r.Partition]; exists {
-			if at.Epoch > r.LeaderEpoch || at.Epoch == r.LeaderEpoch && at.Offset > r.Offset {
+			if set.Less(at) {
 				continue
 			}
 		}
-		toffsets[r.Partition] = EpochOffset{
-			r.LeaderEpoch,
-			r.Offset + 1, // need to advice to next offset to move forward
-		}
+
+		toffsets[r.Partition] = set
 	}
 
 	var rerr error // return error
@@ -2469,11 +2478,11 @@ func (cl *Client) MarkCommitRecords(rs ...*Record) {
 			curTopic = r.Topic
 		}
 
-		current := curPartitions[r.Partition]
+		current, ok := curPartitions[r.Partition]
 		if newHead := (EpochOffset{
 			r.LeaderEpoch,
 			r.Offset + 1,
-		}); current.head.Less(newHead) {
+		}); !ok || current.head.Less(newHead) {
 			curPartitions[r.Partition] = uncommit{
 				dirty:     current.dirty,
 				committed: current.committed,
@@ -2509,8 +2518,8 @@ func (cl *Client) MarkCommitOffsets(unmarked map[string]map[int32]EpochOffset) {
 		}
 
 		for partition, newHead := range partitions {
-			current := curPartitions[partition]
-			if current.head.Less(newHead) {
+			current, ok := curPartitions[partition]
+			if !ok || current.head.Less(newHead) {
 				curPartitions[partition] = uncommit{
 					dirty:     current.dirty,
 					committed: current.committed,

pkg/kgo/metadata.go

@@ -926,8 +926,20 @@ func (cl *Client) mergeTopicPartitions(
 	for _, newTP := range newPartitions {
 		if isProduce && newTP.records.recBufsIdx == -1 {
 			newTP.records.sink.addRecBuf(newTP.records)
+			cl.cfg.logger.Log(LogLevelDebug, "metadata refresh new produce partition",
+				"topic", topic,
+				"partition", newTP.partition(),
+				"leader", newTP.leader,
+				"leader_epoch", newTP.leaderEpoch,
+			)
 		} else if !isProduce && newTP.cursor.cursorsIdx == -1 {
 			newTP.cursor.source.addCursor(newTP.cursor)
+			cl.cfg.logger.Log(LogLevelDebug, "metadata refresh new consume partition",
+				"topic", topic,
+				"partition", newTP.partition(),
+				"leader", newTP.leader,
+				"leader_epoch", newTP.leaderEpoch,
+			)
 		}
 	}
 }

pkg/kgo/producer.go

@@ -624,7 +624,7 @@ func (p *producer) finishPromises(b batchPromise) {
 	}()
 start:
 	for i, pr := range b.recs {
-		pr.LeaderEpoch = 0
+		pr.LeaderEpoch = -1
 		if b.baseOffset == -1 {
 			// if the base offset is invalid/unknown (-1), all record offsets should
 			// be treated as unknown

pkg/kgo/record_and_fetch.go

@@ -130,7 +130,9 @@ type Record struct {
 	ProducerID int64
 
 	// LeaderEpoch is the leader epoch of the broker at the time this
-	// record was written, or -1 if on message sets.
+	// record was written, or -1 if on message sets. When producing,
+	// this is always set to -1 (producers do not use this field and the
+	// broker does not reply with the epoch).
 	//
 	// For committing records, it is not recommended to modify the
 	// LeaderEpoch. Clients use the LeaderEpoch for data loss detection.