kgo consumer: unconditionally move next record ask for record batches
This is an extension of a recent prior commit. While decoding the
records should not fail, if it does, we did not bump the next ask
offset. We do so now while also extending the error to indicate that we
are skipping offsets.
@@ -949,9 +949,20 @@ func (o *cursorOffsetNext) processRecordBatch(
return
}
+ nextAskOffset := lastOffset + 1
+ defer func() {
+ if o.offset < nextAskOffset {
+ // KAFKA-5443: compacted topics preserve the last offset in a
+ // batch, even if the last record is removed, meaning that
+ // using offsets from records alone may not get us to the next
+ // offset we need to ask for.
+ o.offset = nextAskOffset
+ }
+ }()
+
krecords, err := readRawRecords(int(batch.NumRecords), rawRecords)
if err != nil {
- fp.Err = fmt.Errorf("invalid record batch: %v", err)
+ fp.Err = fmt.Errorf("invalid record batch: %v, on offset %d, asking next for offset %d", err, o.offset, nextAskOffset)
return
}
@@ -968,15 +979,6 @@ func (o *cursorOffsetNext) processRecordBatch(
o.maybeKeepRecord(fp, record, abortBatch)
}
- nextAskOffset := lastOffset + 1
- if o.offset < nextAskOffset {
- // KAFKA-5443: compacted topics preserve the last offset in a
- // batch, even if the last record is removed, meaning that
- // using offsets from records alone may not get us to the next
- // offset we need to ask for.
- o.offset = nextAskOffset
- }
-
if abortBatch && lastRecord != nil && lastRecord.Attrs.IsControl() {
aborter.trackAbortedPID(batch.ProducerID)
}
