Merge pull request #846 from rodaine/invalid-base-offset

Fix handling of invalid base offsets

Author: twmb

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

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)