Fix handling of invalid base offsets

twmb · Oct 22, 2024 · 2eed36e · 2eed36e
1 parent cea7aa5
commit 2eed36e
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 2 deletions.
diff --git a/pkg/kgo/producer.go b/pkg/kgo/producer.go
@@ -553,7 +553,13 @@ start:
 	p.promisesMu.Lock()
 	for i, pr := range b.recs {
 		pr.LeaderEpoch = 0
-		pr.Offset = b.baseOffset + int64(i)
+		if b.baseOffset == -1 {
+			// if the base offset is invalid/unknown (-1), all record offsets should
+			// be treated as unknown
+			pr.Offset = -1
+		} else {
+			pr.Offset = b.baseOffset + int64(i)
+		}
 		pr.Partition = b.partition
 		pr.ProducerID = b.pid
 		pr.ProducerEpoch = b.epoch

diff --git a/pkg/kgo/source.go b/pkg/kgo/source.go
@@ -1773,7 +1773,11 @@ func recordToRecord(
 		ProducerID:    batch.ProducerID,
 		ProducerEpoch: batch.ProducerEpoch,
 		LeaderEpoch:   batch.PartitionLeaderEpoch,
-		Offset:        batch.FirstOffset + int64(record.OffsetDelta),
+	}
+	if batch.FirstOffset == -1 {
+		r.Offset = -1
+	} else {
+		r.Offset = batch.FirstOffset + int64(record.OffsetDelta)
 	}
 	if r.Attrs.TimestampType() == 0 {
 		r.Timestamp = timeFromMillis(batch.FirstTimestamp + record.TimestampDelta64)