diff --git a/batch.go b/batch.go index eb742712..5c94dd14 100644 --- a/batch.go +++ b/batch.go @@ -278,17 +278,33 @@ func (batch *Batch) readMessage( // - `batch.err` for a "success" from the previous timeout check // - `batch.msgs.lengthRemain` to ensure that this EOF is not due // to MaxBytes truncation - // - `batch.lastOffset` to ensure that the message format contains - // `lastOffset` - if errors.Is(batch.err, io.EOF) && batch.msgs.lengthRemain == 0 && batch.lastOffset != -1 { - // Log compaction can create batches that end with compacted - // records so the normal strategy that increments the "next" - // offset as records are read doesn't work as the compacted - // records are "missing" and never get "read". + if errors.Is(batch.err, io.EOF) && batch.msgs.lengthRemain == 0 { + // Two things leave offsets behind that no returned message + // accounted for, and both are resolved by resuming past the + // highest offset the reader is known to have consumed: // - // In order to reliably reach the next non-compacted offset we - // jump past the saved lastOffset. - batch.offset = batch.lastOffset + 1 + // - Log compaction can create batches that end with compacted + // records so the normal strategy that increments the "next" + // offset as records are read doesn't work as the compacted + // records are "missing" and never get "read". In order to + // reliably reach the next non-compacted offset we jump past + // the saved lastOffset, which is -1 when the message format + // does not carry one. + // - Control batches and batches belonging to aborted + // transactions are consumed without being returned. Leaving + // their offsets behind would mean fetching the same batches + // again and never making progress. + // + // The offset only ever moves forward here: a response that + // returned no message at all must not rewind the partition. + next := batch.offset + if batch.lastOffset != -1 && batch.lastOffset+1 > next { + next = batch.lastOffset + 1 + } + if skipped := batch.msgs.lastSkippedOffset; skipped+1 > next { + next = skipped + 1 + } + batch.offset = next } } default: diff --git a/conn.go b/conn.go index 9f9f2590..12c753d4 100644 --- a/conn.go +++ b/conn.go @@ -114,6 +114,9 @@ type ReadBatchConfig struct { // IsolationLevel controls the visibility of transactional records. // ReadUncommitted makes all records visible. With ReadCommitted only // non-transactional and committed records are visible. + // + // Defaults to ReadUncommitted, matching Kafka's own default. See the + // IsolationLevel constants for the trade-off ReadCommitted carries. IsolationLevel IsolationLevel // MaxWait is the amount of time for the broker while waiting to hit the @@ -125,11 +128,33 @@ type ReadBatchConfig struct { MaxWait time.Duration } +// IsolationLevel controls which transactional records a consumer is shown. +// +// Transaction markers are hidden at both levels: they are bookkeeping written +// by the transaction coordinator, not records, and the protocol requires that +// clients never surface them. type IsolationLevel int8 const ( + // ReadUncommitted returns every record, including those written by a + // transaction that has not committed and those written by one that + // aborted. This is the zero value, and matches Kafka's own default. ReadUncommitted IsolationLevel = 0 - ReadCommitted IsolationLevel = 1 + + // ReadCommitted returns only non-transactional records and records from + // committed transactions. + // + // Two things are worth knowing before choosing it: + // + // It requires a broker supporting version 4 or above of the Fetch API + // (Kafka 0.11 and later). Against an older broker the request carries no + // isolation level and the setting has no effect. + // + // Reads stop at the last stable offset rather than the high watermark, so + // a producer that leaves a transaction open blocks the consumer behind it + // no matter how many records were committed after it. That is inherent to + // how Kafka implements the guarantee, and applies to every client. + ReadCommitted IsolationLevel = 1 ) var ( @@ -848,14 +873,15 @@ func (c *Conn) ReadBatchWith(cfg ReadBatchConfig) *Batch { var throttle int32 var highWaterMark int64 var remain int + var aborted []abortedTransaction switch fetchVersion { case v10: - throttle, highWaterMark, remain, err = readFetchResponseHeaderV10(&c.rbuf, size) + throttle, highWaterMark, remain, aborted, err = readFetchResponseHeaderV10(&c.rbuf, size) case v5: - throttle, highWaterMark, remain, err = readFetchResponseHeaderV5(&c.rbuf, size) + throttle, highWaterMark, remain, aborted, err = readFetchResponseHeaderV5(&c.rbuf, size) default: - throttle, highWaterMark, remain, err = readFetchResponseHeaderV2(&c.rbuf, size) + throttle, highWaterMark, remain, aborted, err = readFetchResponseHeaderV2(&c.rbuf, size) } if errors.Is(err, errShortRead) { err = checkTimeoutErr(adjustedDeadline) @@ -864,9 +890,9 @@ func (c *Conn) ReadBatchWith(cfg ReadBatchConfig) *Batch { var msgs *messageSetReader if err == nil { if highWaterMark == offset { - msgs = &messageSetReader{empty: true} + msgs = &messageSetReader{empty: true, lastSkippedOffset: -1} } else { - msgs, err = newMessageSetReader(&c.rbuf, remain) + msgs, err = newMessageSetReader(&c.rbuf, remain, aborted) } } if errors.Is(err, errShortRead) { diff --git a/message_reader.go b/message_reader.go index a0a0385e..d4f09369 100644 --- a/message_reader.go +++ b/message_reader.go @@ -3,9 +3,11 @@ package kafka import ( "bufio" "bytes" + "encoding/binary" "fmt" "io" "log" + "sort" ) type readBytesFunc func(*bufio.Reader, int, int) (int, error) @@ -23,6 +25,25 @@ type messageSetReader struct { lengthRemain int decompressed *bytes.Buffer + + // abortedTxns holds the transactions the broker reported as aborted in the + // fetch response, ordered by first offset and consumed front to back as the + // response is read. abortedProducers holds the producers whose records are + // currently being dropped: a producer enters the set when the reader + // reaches the first offset of one of its aborted transactions, and leaves + // it again when the matching abort marker is read. + // + // Brokers only send the list when the fetch asked for ReadCommitted, so + // under ReadUncommitted both stay empty and nothing is dropped. + abortedTxns []abortedTransaction + abortedProducers map[int64]struct{} + + // lastSkippedOffset is the highest offset consumed by a batch that was + // hidden from the caller, or -1 if there was none. Those offsets appear in + // no message, so Batch needs this to resume after them; without it a + // response ending in a control batch would leave the offset behind the + // marker and the same batch would be fetched forever. + lastSkippedOffset int64 } type readerStack struct { @@ -59,6 +80,32 @@ type messagesHeader struct { } } +// control returns true if the header describes a control batch. Control +// batches hold transaction markers written by the transaction coordinator +// rather than records written by a producer, and the protocol requires that +// they are not exposed to the application. Consumers that do surface them see +// an empty message for every committed transaction. +// See https://kafka.apache.org/documentation/#controlbatch +// +// Only v2 message sets support transactions, so v0 and v1 are never control +// batches. +func (h messagesHeader) control() bool { + const controlMask = 0x20 + return h.magic == 2 && (h.v2.attributes&controlMask) != 0 +} + +// transactional returns true if the header describes a batch that a producer +// wrote inside a transaction. Such a batch is only visible to a ReadCommitted +// consumer once the transaction commits, so the reader has to be able to tell +// it apart from an ordinary one. +// +// Only v2 message sets support transactions, so v0 and v1 are never +// transactional. +func (h messagesHeader) transactional() bool { + const transactionalMask = 0x10 + return h.magic == 2 && (h.v2.attributes&transactionalMask) != 0 +} + func (h messagesHeader) compression() (codec CompressionCodec, err error) { const compressionCodecMask = 0x07 var code int8 @@ -81,13 +128,24 @@ func (h messagesHeader) badMagic() error { return fmt.Errorf("unsupported magic byte %d in header", h.magic) } -func newMessageSetReader(reader *bufio.Reader, remain int) (*messageSetReader, error) { +// newMessageSetReader constructs a reader over the message set of a fetch +// response. aborted is the list of aborted transactions the broker returned +// alongside it, which may be nil; the reader uses it to drop the records those +// transactions produced. +func newMessageSetReader(reader *bufio.Reader, remain int, aborted []abortedTransaction) (*messageSetReader, error) { + // Brokers return the list in offset order, but consuming it front to back + // depends on that, so sort rather than assume. + sort.Slice(aborted, func(i, j int) bool { + return aborted[i].FirstOffset < aborted[j].FirstOffset + }) res := &messageSetReader{ readerStack: &readerStack{ reader: reader, remain: remain, }, - decompressed: acquireBuffer(), + decompressed: acquireBuffer(), + abortedTxns: aborted, + lastSkippedOffset: -1, } err := res.readHeader() return res, err @@ -122,26 +180,188 @@ func (r *messageSetReader) discard() (err error) { func (r *messageSetReader) readMessage(min int64, key readBytesFunc, val readBytesFunc) ( offset int64, lastOffset int64, timestamp int64, headers []Header, err error) { - if r.empty { - err = RequestTimedOut + for { + if r.empty { + err = RequestTimedOut + return + } + if err = r.readHeader(); err != nil { + return + } + switch r.header.magic { + case 0, 1: + offset, timestamp, headers, err = r.readMessageV1(min, key, val) + // Set an invalid value so that it can be ignored + lastOffset = -1 + case 2: + // Two kinds of batch are consumed here instead of being returned: + // control batches, which hold transaction markers rather than + // messages, and batches whose transaction the broker told us was + // aborted. Neither may be exposed to the application, so keep + // reading until a regular batch or the end of the response. + r.consumeAbortedTransactionsUpTo(r.header.firstOffset) + if r.count == 0 { + // A batch the log cleaner emptied. readHeader consumed it + // whole and accounted for its offsets; dispatching on its + // header would consume records of the batch that follows. + continue + } + switch { + case r.header.control(): + if err = r.skipControlRecordV2(min); err != nil { + return + } + continue + case r.abortedBatch(): + if err = r.skipBatchV2(); err != nil { + return + } + continue + } + offset, lastOffset, timestamp, headers, err = r.readMessageV2(min, key, val) + default: + err = r.header.badMagic() + } return } - if err = r.readHeader(); err != nil { +} + +// consumeAbortedTransactionsUpTo moves every aborted transaction that starts at +// or before offset into the set of producers whose records are being dropped. +// A producer stays in that set until its abort marker is read, which is what +// makes the records between the two invisible. +// +// The reader calls this with the first offset of each batch before deciding +// what to do with it. Repeat calls for the records of one batch are harmless: +// every entry it could match has already been removed from the list. +func (r *messageSetReader) consumeAbortedTransactionsUpTo(offset int64) { + for len(r.abortedTxns) > 0 && r.abortedTxns[0].FirstOffset <= offset { + if r.abortedProducers == nil { + r.abortedProducers = make(map[int64]struct{}) + } + r.abortedProducers[r.abortedTxns[0].ProducerID] = struct{}{} + r.abortedTxns = r.abortedTxns[1:] + } +} + +// abortedBatch reports whether the records of the current batch belong to a +// transaction the broker reported as aborted. Batches written outside a +// transaction are never dropped, even when they sit between two that were. +func (r *messageSetReader) abortedBatch() bool { + if len(r.abortedProducers) == 0 || !r.header.transactional() { + return false + } + _, aborted := r.abortedProducers[r.header.v2.producerID] + return aborted +} + +// skipControlRecordV2 reads a record from a control batch and discards it. +// discardN satisfies readBytesFunc, so the record is consumed by the same code +// path as a regular one and the reader's bookkeeping is left unchanged. The key +// is kept to identify the marker, which decides both what is logged and whether +// a producer stops being a reason to drop records. +func (r *messageSetReader) skipControlRecordV2(min int64) (err error) { + // Read the header before consuming the record: doing so can exhaust and pop + // a reader stack, replacing r.header with the enclosing one. + producerID := r.header.v2.producerID + lastOffset := r.batchLastOffset() + + var key []byte + captureKey := func(br *bufio.Reader, size int, nbytes int) (remain int, err error) { + key, remain, err = readNewBytes(br, size, nbytes) return } - switch r.header.magic { - case 0, 1: - offset, timestamp, headers, err = r.readMessageV1(min, key, val) - // Set an invalid value so that it can be ignored - lastOffset = -1 - case 2: - offset, lastOffset, timestamp, headers, err = r.readMessageV2(min, key, val) - default: - err = r.header.badMagic() + if _, _, _, _, err = r.readMessageV2(min, captureKey, discardN); err != nil { + return + } + + marker := controlRecordType(key) + if marker == controlRecordAbort { + // The transaction this producer opened is over, so its records stop + // being dropped. A commit marker needs no equivalent: a producer that + // committed was never added to the set in the first place. + delete(r.abortedProducers, producerID) + } + r.noteBatchSkipped(lastOffset) + + if r.debug { + r.log("Skipped %s control record for producerID=%d", + controlRecordTypeName(marker), producerID) } return } +// skipBatchV2 discards the record section of the current batch without decoding +// it, for a batch whose transaction was aborted. +// +// lengthRemain is the batch length minus its 49 header bytes, which is exactly +// that record section, so a compressed batch is discarded in its compressed +// form and never inflated. That only holds before any record of the batch has +// been read: once readMessageV2 has started on a compressed batch it has pushed +// a reader stack holding the decompressed bytes and the accounting no longer +// lines up. +func (r *messageSetReader) skipBatchV2() (err error) { + if r.count != int(r.header.v2.count) { + return fmt.Errorf("skipBatchV2 called after %d of %d records were read", + int(r.header.v2.count)-r.count, r.header.v2.count) + } + lastOffset := r.batchLastOffset() + if err = r.discardN(r.lengthRemain); err != nil { + return + } + if r.debug { + r.log("Skipped aborted batch of %d records for producerID=%d", + r.header.v2.count, r.header.v2.producerID) + } + r.noteBatchSkipped(lastOffset) + r.lengthRemain = 0 + r.count = 0 + r.unwindStack() + return +} + +// batchLastOffset returns the offset of the last record of the current v2 +// batch. +func (r *messageSetReader) batchLastOffset() int64 { + return r.header.firstOffset + int64(r.header.v2.lastOffsetDelta) +} + +// noteBatchSkipped records that every offset up to lastOffset has been consumed +// by a batch the caller never sees, so that Batch can resume past it. +func (r *messageSetReader) noteBatchSkipped(lastOffset int64) { + if lastOffset > r.lastSkippedOffset { + r.lastSkippedOffset = lastOffset + } +} + +// Marker types held in the int16 type field of a control record's key. +const ( + controlRecordAbort int16 = 0 + controlRecordCommit int16 = 1 + controlRecordUnknown int16 = -1 +) + +// controlRecordType returns the type of the marker held in a control record's +// key, which is an int16 version followed by an int16 type. A key of any other +// shape is not a marker this client understands. +func controlRecordType(key []byte) int16 { + if len(key) != 4 { + return controlRecordUnknown + } + return int16(binary.BigEndian.Uint16(key[2:])) +} + +func controlRecordTypeName(t int16) string { + switch t { + case controlRecordAbort: + return "ABORT" + case controlRecordCommit: + return "COMMIT" + default: + return "unknown" + } +} + func (r *messageSetReader) readMessageV1(min int64, key readBytesFunc, val readBytesFunc) ( offset int64, timestamp int64, headers []Header, err error) { @@ -482,6 +702,21 @@ func (r *messageSetReader) readHeader() (err error) { r.count = int(r.header.v2.count) // Subtracts the header bytes from the length r.lengthRemain = int(r.header.length) - 49 + // The log cleaner can remove every record a batch held and retain the + // batch itself to preserve the producer's state, so a batch with no + // records at all is a normal sight on a compacted topic. Reading its + // header consumed the whole batch, and the next call replaces the + // header, so the offsets it spans have to be accounted for now: no + // message will ever surface them. + if r.count == 0 { + r.noteBatchSkipped(r.batchLastOffset()) + if r.lengthRemain > 0 { + if err = r.discardN(r.lengthRemain); err != nil { + return + } + } + r.lengthRemain = 0 + } if r.debug { r.log("Read v2 header with count=%d offset=%d len=%d magic=%d attributes=%d", r.count, r.header.firstOffset, r.header.length, r.header.magic, r.header.v2.attributes) } diff --git a/message_reader_control_test.go b/message_reader_control_test.go new file mode 100644 index 00000000..86c8f7cf --- /dev/null +++ b/message_reader_control_test.go @@ -0,0 +1,699 @@ +package kafka + +import ( + "bufio" + "bytes" + "encoding/binary" + "errors" + "io" + "testing" + "time" + + "github.com/segmentio/kafka-go/compress/gzip" +) + +// appendZigZagVarInt encodes v the way the record format does, matching +// readVarInt's decoding. +func appendZigZagVarInt(b []byte, v int64) []byte { + u := uint64(v<<1) ^ uint64(v>>63) + for u >= 0x80 { + b = append(b, byte(u)|0x80) + u >>= 7 + } + return append(b, byte(u)) +} + +func appendUint16(b []byte, v uint16) []byte { + var buf [2]byte + binary.BigEndian.PutUint16(buf[:], v) + return append(b, buf[:]...) +} + +func appendUint32(b []byte, v uint32) []byte { + var buf [4]byte + binary.BigEndian.PutUint32(buf[:], v) + return append(b, buf[:]...) +} + +func appendUint64(b []byte, v uint64) []byte { + var buf [8]byte + binary.BigEndian.PutUint64(buf[:], v) + return append(b, buf[:]...) +} + +// v2Record builds a single record of the v2 message format. +func v2Record(offsetDelta int64, key, value []byte) []byte { + var body []byte + body = append(body, 0) // attributes + body = appendZigZagVarInt(body, 0) + body = appendZigZagVarInt(body, offsetDelta) + body = appendZigZagVarInt(body, int64(len(key))) + body = append(body, key...) + body = appendZigZagVarInt(body, int64(len(value))) + body = append(body, value...) + body = appendZigZagVarInt(body, 0) // header count + + out := appendZigZagVarInt(nil, int64(len(body))) + return append(out, body...) +} + +// batchOpts describes the header fields of a v2 record batch that the reader's +// transaction handling depends on. +type batchOpts struct { + firstOffset int64 + producerID int64 + control bool + transactional bool + codec CompressionCodec // nil leaves the records uncompressed +} + +// v2Batch builds a record batch of the v2 message format. The CRC is left zero +// because the reader does not verify it. +func v2Batch(t *testing.T, o batchOpts, records ...[]byte) []byte { + t.Helper() + + var recs []byte + for _, r := range records { + recs = append(recs, r...) + } + + var attributes uint16 + if o.control { + attributes |= 0x20 + } + if o.transactional { + attributes |= 0x10 + } + if o.codec != nil { + attributes |= uint16(o.codec.Code()) & 0x07 + + buf := &bytes.Buffer{} + w := o.codec.NewWriter(buf) + if _, err := w.Write(recs); err != nil { + t.Fatalf("compressing records: %v", err) + } + if err := w.Close(); err != nil { + t.Fatalf("closing the compression writer: %v", err) + } + recs = buf.Bytes() + } + + b := make([]byte, 0, 61+len(recs)) + b = appendUint64(b, uint64(o.firstOffset)) + b = appendUint32(b, uint32(49+len(recs))) // length after this field + b = appendUint32(b, 0) // partitionLeaderEpoch + b = append(b, 2) // magic + b = appendUint32(b, 0) // crc + b = appendUint16(b, attributes) + b = appendUint32(b, uint32(len(records)-1)) // lastOffsetDelta + b = appendUint64(b, 0) // firstTimestamp + b = appendUint64(b, 0) // maxTimestamp + b = appendUint64(b, uint64(o.producerID)) + b = appendUint16(b, 0) // producerEpoch + b = appendUint32(b, 0) // baseSequence + b = appendUint32(b, uint32(len(records))) // record count + return append(b, recs...) +} + +// marker builds the record a transaction coordinator writes to close a +// transaction: a key of int16 version and int16 type, and a value of int16 +// version and int32 coordinator epoch. +func marker(offsetDelta int64, markerType int16) []byte { + key := make([]byte, 4) + binary.BigEndian.PutUint16(key[2:], uint16(markerType)) + return v2Record(offsetDelta, key, []byte{0x00, 0x00, 0x00, 0x00, 0x00, 0x00}) +} + +// controlBatch builds the single-record control batch that closes a +// transaction at offset. +func controlBatch(t *testing.T, offset, producerID int64, markerType int16) []byte { + t.Helper() + return v2Batch(t, + batchOpts{firstOffset: offset, producerID: producerID, control: true, transactional: true}, + marker(0, markerType)) +} + +// emptyBatch builds a record batch whose records the log cleaner removed. The +// cleaner keeps the original offset range when it retains an empty batch, so +// lastOffsetDelta is the one the batch had before cleaning, not the -1 that +// zero records would otherwise imply. +func emptyBatch(t *testing.T, o batchOpts, lastOffsetDelta int32) []byte { + t.Helper() + b := v2Batch(t, o) + binary.BigEndian.PutUint32(b[23:27], uint32(lastOffsetDelta)) + return b +} + +func newControlTestReader(t *testing.T, data []byte, aborted ...abortedTransaction) *messageSetReader { + t.Helper() + r, err := newMessageSetReader(bufio.NewReader(bytes.NewReader(data)), len(data), aborted) + if err != nil { + t.Fatalf("newMessageSetReader: %v", err) + } + return r +} + +func readOneMessage(r *messageSetReader, offset int64) (msg Message, err error) { + keyFunc := func(r *bufio.Reader, size int, nbytes int) (remain int, err error) { + msg.Key, remain, err = readNewBytes(r, size, nbytes) + return + } + valFunc := func(r *bufio.Reader, size int, nbytes int) (remain int, err error) { + msg.Value, remain, err = readNewBytes(r, size, nbytes) + return + } + msg.Offset, _, _, _, err = r.readMessage(offset, keyFunc, valFunc) + return +} + +// readAllValues drains the reader and returns the value of every message it +// surfaced. Any error other than the exhaustion of the response fails the test. +func readAllValues(t *testing.T, r *messageSetReader) []string { + t.Helper() + + var values []string + var offset int64 + for { + msg, err := readOneMessage(r, offset) + if err != nil { + if !errors.Is(err, errShortRead) { + t.Fatalf("readMessage after %d messages: %v", len(values), err) + } + return values + } + values = append(values, string(msg.Value)) + offset = msg.Offset + 1 + } +} + +// TestReadMessageSkipsLeadingControlBatch verifies that a control batch ahead +// of the data does not surface as a message. +func TestReadMessageSkipsLeadingControlBatch(t *testing.T) { + data := append( + controlBatch(t, 0, 0, controlRecordCommit), + v2Batch(t, batchOpts{firstOffset: 1}, v2Record(0, []byte("key-1"), []byte("value-1")))..., + ) + r := newControlTestReader(t, data) + + msg, err := readOneMessage(r, 0) + if err != nil { + t.Fatalf("readMessage: %v", err) + } + if string(msg.Key) != "key-1" || string(msg.Value) != "value-1" { + t.Errorf("expected key-1/value-1, got key=%q value=%q", msg.Key, msg.Value) + } + if msg.Offset != 1 { + t.Errorf("offset = %d, want 1", msg.Offset) + } +} + +// TestReadMessageSkipsInterleavedControlBatches covers the usual layout of a +// transactional producer, which writes a marker after every committed batch. +func TestReadMessageSkipsInterleavedControlBatches(t *testing.T) { + var data []byte + data = append(data, v2Batch(t, batchOpts{firstOffset: 0}, v2Record(0, []byte("k0"), []byte("v0")))...) + data = append(data, controlBatch(t, 1, 0, controlRecordCommit)...) + data = append(data, v2Batch(t, batchOpts{firstOffset: 2}, v2Record(0, []byte("k1"), []byte("v1")))...) + data = append(data, controlBatch(t, 3, 0, controlRecordCommit)...) + data = append(data, v2Batch(t, batchOpts{firstOffset: 4}, v2Record(0, []byte("k2"), []byte("v2")))...) + + r := newControlTestReader(t, data) + + if got, want := readAllValues(t, r), []string{"v0", "v1", "v2"}; !equalStrings(got, want) { + t.Errorf("values = %q, want %q", got, want) + } +} + +// TestReadMessageTrailingControlBatchIsNotAMessage verifies that a response +// ending in a control batch reports exhaustion instead of returning an empty +// message. errShortRead is how the reader signals an exhausted response, and +// Batch.readMessage already translates it into end of batch. +func TestReadMessageTrailingControlBatchIsNotAMessage(t *testing.T) { + data := append( + v2Batch(t, batchOpts{firstOffset: 0}, v2Record(0, []byte("k0"), []byte("v0"))), + controlBatch(t, 1, 0, controlRecordCommit)..., + ) + r := newControlTestReader(t, data) + + msg, err := readOneMessage(r, 0) + if err != nil { + t.Fatalf("readMessage: %v", err) + } + if string(msg.Value) != "v0" { + t.Fatalf("first value = %q, want v0", msg.Value) + } + + msg, err = readOneMessage(r, msg.Offset+1) + if err == nil { + t.Fatalf("expected an error, got key=%q value=%q", msg.Key, msg.Value) + } + if !errors.Is(err, errShortRead) { + t.Logf("second readMessage error = %v (any non-nil error ends the batch)", err) + } +} + +// TestControlHeaderPredicate covers the attribute bit itself. +func TestControlHeaderPredicate(t *testing.T) { + var h messagesHeader + h.magic = 2 + h.v2.attributes = 0x20 + if !h.control() { + t.Error("control() = false for a v2 header with the control bit set") + } + + h.v2.attributes = 0x21 // control + gzip + if !h.control() { + t.Error("control() = false when the control bit is set alongside compression") + } + + h.v2.attributes = 0x00 + if h.control() { + t.Error("control() = true for an ordinary v2 batch") + } + + // v0 and v1 have no transactions, so the bit is not meaningful there. + h.magic = 1 + h.v2.attributes = 0x20 + if h.control() { + t.Error("control() = true for a v1 header") + } +} + +// TestTransactionalHeaderPredicate covers the transactional attribute bit, +// which is what keeps the aborted-transaction filter away from batches written +// outside a transaction. +func TestTransactionalHeaderPredicate(t *testing.T) { + var h messagesHeader + h.magic = 2 + + h.v2.attributes = 0x10 + if !h.transactional() { + t.Error("transactional() = false for a v2 header with the transactional bit set") + } + + h.v2.attributes = 0x30 // transactional + control, as a marker batch carries + if !h.transactional() { + t.Error("transactional() = false when the control bit is set alongside it") + } + + h.v2.attributes = 0x20 // control only + if h.transactional() { + t.Error("transactional() = true for a batch that only set the control bit") + } + + h.v2.attributes = 0x00 + if h.transactional() { + t.Error("transactional() = true for an ordinary v2 batch") + } + + h.magic = 1 + h.v2.attributes = 0x10 + if h.transactional() { + t.Error("transactional() = true for a v1 header") + } +} + +// TestControlRecordType covers the marker type decoded from a control record's +// key, which decides whether a producer stops being a reason to drop records. +func TestControlRecordType(t *testing.T) { + tests := []struct { + name string + key []byte + want int16 + }{ + {"abort", []byte{0x00, 0x00, 0x00, 0x00}, controlRecordAbort}, + {"commit", []byte{0x00, 0x00, 0x00, 0x01}, controlRecordCommit}, + {"unrecognized type", []byte{0x00, 0x00, 0x00, 0x09}, 9}, + {"wrong key size", []byte{0x00, 0x00}, controlRecordUnknown}, + {"nil key", nil, controlRecordUnknown}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + if got := controlRecordType(test.key); got != test.want { + t.Errorf("controlRecordType(%x) = %d, want %d", test.key, got, test.want) + } + }) + } + + if got := controlRecordTypeName(controlRecordAbort); got != "ABORT" { + t.Errorf("name of an abort marker = %q, want ABORT", got) + } + if got := controlRecordTypeName(controlRecordCommit); got != "COMMIT" { + t.Errorf("name of a commit marker = %q, want COMMIT", got) + } + if got := controlRecordTypeName(controlRecordUnknown); got != "unknown" { + t.Errorf("name of an unrecognized marker = %q, want unknown", got) + } +} + +// TestReadMessageSkipsAbortedTransaction covers the case issue #1332 reports: a +// ReadCommitted consumer must not see the records of a transaction the broker +// listed as aborted. +func TestReadMessageSkipsAbortedTransaction(t *testing.T) { + var data []byte + data = append(data, v2Batch(t, + batchOpts{firstOffset: 0, producerID: 7, transactional: true}, + v2Record(0, []byte("k0"), []byte("aborted-0")), + v2Record(1, []byte("k1"), []byte("aborted-1")))...) + data = append(data, controlBatch(t, 2, 7, controlRecordAbort)...) + data = append(data, v2Batch(t, + batchOpts{firstOffset: 3, producerID: 8, transactional: true}, + v2Record(0, []byte("k2"), []byte("committed")))...) + data = append(data, controlBatch(t, 4, 8, controlRecordCommit)...) + + r := newControlTestReader(t, data, abortedTransaction{ProducerID: 7, FirstOffset: 0}) + + if got, want := readAllValues(t, r), []string{"committed"}; !equalStrings(got, want) { + t.Errorf("values = %q, want %q", got, want) + } +} + +// TestReadMessageKeepsNonTransactionalRecordsWhileFiltering verifies that a +// producer writing outside a transaction is unaffected by another producer's +// abort, even when its batches are interleaved with the aborted ones. +func TestReadMessageKeepsNonTransactionalRecordsWhileFiltering(t *testing.T) { + var data []byte + data = append(data, v2Batch(t, + batchOpts{firstOffset: 0, producerID: 7, transactional: true}, + v2Record(0, []byte("k0"), []byte("aborted")))...) + data = append(data, v2Batch(t, + batchOpts{firstOffset: 1, producerID: 9}, + v2Record(0, []byte("k1"), []byte("plain")))...) + data = append(data, controlBatch(t, 2, 7, controlRecordAbort)...) + + r := newControlTestReader(t, data, abortedTransaction{ProducerID: 7, FirstOffset: 0}) + + if got, want := readAllValues(t, r), []string{"plain"}; !equalStrings(got, want) { + t.Errorf("values = %q, want %q", got, want) + } +} + +// TestReadMessageResumesProducerAfterAbortMarker is the test that pins the +// removal of a producer from the aborted set. A producer that aborts one +// transaction can immediately commit the next one, and those later records must +// be delivered. An implementation that never clears the set passes every other +// test here and fails this one. +func TestReadMessageResumesProducerAfterAbortMarker(t *testing.T) { + var data []byte + data = append(data, v2Batch(t, + batchOpts{firstOffset: 0, producerID: 7, transactional: true}, + v2Record(0, []byte("k0"), []byte("aborted")))...) + data = append(data, controlBatch(t, 1, 7, controlRecordAbort)...) + data = append(data, v2Batch(t, + batchOpts{firstOffset: 2, producerID: 7, transactional: true}, + v2Record(0, []byte("k1"), []byte("committed")))...) + data = append(data, controlBatch(t, 3, 7, controlRecordCommit)...) + + r := newControlTestReader(t, data, abortedTransaction{ProducerID: 7, FirstOffset: 0}) + + if got, want := readAllValues(t, r), []string{"committed"}; !equalStrings(got, want) { + t.Errorf("values = %q, want %q", got, want) + } +} + +// TestReadMessageStartsFilteringAtTheTransactionFirstOffset verifies that an +// aborted transaction only hides records from its own first offset onwards. +// Earlier records from the same producer belong to a transaction that already +// committed. +func TestReadMessageStartsFilteringAtTheTransactionFirstOffset(t *testing.T) { + var data []byte + data = append(data, v2Batch(t, + batchOpts{firstOffset: 0, producerID: 7, transactional: true}, + v2Record(0, []byte("k0"), []byte("earlier-commit")))...) + data = append(data, controlBatch(t, 1, 7, controlRecordCommit)...) + data = append(data, v2Batch(t, + batchOpts{firstOffset: 2, producerID: 7, transactional: true}, + v2Record(0, []byte("k1"), []byte("aborted")))...) + data = append(data, controlBatch(t, 3, 7, controlRecordAbort)...) + + r := newControlTestReader(t, data, abortedTransaction{ProducerID: 7, FirstOffset: 2}) + + if got, want := readAllValues(t, r), []string{"earlier-commit"}; !equalStrings(got, want) { + t.Errorf("values = %q, want %q", got, want) + } +} + +// TestReadMessageDeliversTransactionalRecordsWithoutAnAbortList covers +// ReadUncommitted, where the broker sends no aborted transaction list. Nothing +// may be dropped then, markers aside. +func TestReadMessageDeliversTransactionalRecordsWithoutAnAbortList(t *testing.T) { + var data []byte + data = append(data, v2Batch(t, + batchOpts{firstOffset: 0, producerID: 7, transactional: true}, + v2Record(0, []byte("k0"), []byte("v0")))...) + data = append(data, controlBatch(t, 1, 7, controlRecordAbort)...) + + r := newControlTestReader(t, data) + + if got, want := readAllValues(t, r), []string{"v0"}; !equalStrings(got, want) { + t.Errorf("values = %q, want %q", got, want) + } +} + +// TestReadMessageSkipsAbortedCompressedBatch verifies that an aborted batch is +// discarded in its compressed form. skipBatchV2 relies on lengthRemain covering +// exactly the record section, which is what makes that possible; getting it +// wrong desynchronizes the reader and corrupts every batch that follows. +func TestReadMessageSkipsAbortedCompressedBatch(t *testing.T) { + var data []byte + data = append(data, v2Batch(t, + batchOpts{firstOffset: 0, producerID: 7, transactional: true, codec: new(gzip.Codec)}, + v2Record(0, []byte("k0"), []byte("aborted-0")), + v2Record(1, []byte("k1"), []byte("aborted-1")))...) + data = append(data, controlBatch(t, 2, 7, controlRecordAbort)...) + // A second producer, so that reaching the committed records does not also + // depend on producer 7 leaving the aborted set. + data = append(data, v2Batch(t, + batchOpts{firstOffset: 3, producerID: 8, transactional: true, codec: new(gzip.Codec)}, + v2Record(0, []byte("k2"), []byte("committed")))...) + data = append(data, controlBatch(t, 4, 8, controlRecordCommit)...) + + r := newControlTestReader(t, data, abortedTransaction{ProducerID: 7, FirstOffset: 0}) + + if got, want := readAllValues(t, r), []string{"committed"}; !equalStrings(got, want) { + t.Errorf("values = %q, want %q", got, want) + } +} + +// TestReadMessageSortsAbortedTransactions verifies that a list arriving out of +// offset order is still consumed correctly. Brokers send it sorted, but the +// front-to-back consumption depends on it, so the reader sorts rather than +// trusts. +func TestReadMessageSortsAbortedTransactions(t *testing.T) { + var data []byte + data = append(data, v2Batch(t, + batchOpts{firstOffset: 0, producerID: 7, transactional: true}, + v2Record(0, []byte("k0"), []byte("aborted-7")))...) + data = append(data, controlBatch(t, 1, 7, controlRecordAbort)...) + data = append(data, v2Batch(t, + batchOpts{firstOffset: 2, producerID: 8, transactional: true}, + v2Record(0, []byte("k1"), []byte("aborted-8")))...) + data = append(data, controlBatch(t, 3, 8, controlRecordAbort)...) + data = append(data, v2Batch(t, + batchOpts{firstOffset: 4, producerID: 9, transactional: true}, + v2Record(0, []byte("k2"), []byte("committed")))...) + data = append(data, controlBatch(t, 5, 9, controlRecordCommit)...) + + r := newControlTestReader(t, data, + abortedTransaction{ProducerID: 8, FirstOffset: 2}, + abortedTransaction{ProducerID: 7, FirstOffset: 0}) + + if got, want := readAllValues(t, r), []string{"committed"}; !equalStrings(got, want) { + t.Errorf("values = %q, want %q", got, want) + } +} + +// TestReadMessageSkipsEmptyControlBatchBetweenData covers a marker batch the +// log cleaner emptied. The cleaner can remove every record a batch held and +// retain the batch itself to preserve the producer's state, so a compacted +// transactional topic serves batches whose record count is zero. Such a batch +// is consumed entirely by its header read, which breaks the assumption that +// the header just inspected belongs to the record about to be read: a control +// check that then consumes "its" record is really discarding the first record +// of the batch that follows. +func TestReadMessageSkipsEmptyControlBatchBetweenData(t *testing.T) { + var data []byte + data = append(data, v2Batch(t, batchOpts{firstOffset: 0}, v2Record(0, []byte("k0"), []byte("v0")))...) + data = append(data, emptyBatch(t, batchOpts{firstOffset: 1, producerID: 7, control: true, transactional: true}, 0)...) + data = append(data, v2Batch(t, batchOpts{firstOffset: 2}, v2Record(0, []byte("k1"), []byte("v1")))...) + + r := newControlTestReader(t, data) + + if got, want := readAllValues(t, r), []string{"v0", "v1"}; !equalStrings(got, want) { + t.Errorf("values = %q, want %q", got, want) + } +} + +// TestReadMessageHidesMarkerAfterEmptyBatch is the same header-ownership bug +// from the other side: an emptied data batch ahead of a marker means the +// control check runs against the empty batch's header, and the marker behind +// it is delivered as a message. +func TestReadMessageHidesMarkerAfterEmptyBatch(t *testing.T) { + var data []byte + data = append(data, v2Batch(t, batchOpts{firstOffset: 0}, v2Record(0, []byte("k0"), []byte("v0")))...) + data = append(data, emptyBatch(t, batchOpts{firstOffset: 1, producerID: 7, transactional: true}, 0)...) + data = append(data, controlBatch(t, 2, 7, controlRecordCommit)...) + data = append(data, v2Batch(t, batchOpts{firstOffset: 3}, v2Record(0, []byte("k1"), []byte("v1")))...) + + r := newControlTestReader(t, data) + + if got, want := readAllValues(t, r), []string{"v0", "v1"}; !equalStrings(got, want) { + t.Errorf("values = %q, want %q", got, want) + } +} + +// newTestBatch wraps a response body in a Batch without a connection, which is +// enough to exercise the offset bookkeeping. The deadline is in the future so +// that exhausting the response reports io.EOF rather than a timeout. +func newTestBatch(t *testing.T, offset int64, data []byte, aborted ...abortedTransaction) *Batch { + t.Helper() + return &Batch{ + msgs: newControlTestReader(t, data, aborted...), + offset: offset, + deadline: time.Now().Add(time.Minute), + } +} + +// drainBatch reads the batch to exhaustion and returns the values it produced +// along with the offset the next fetch would resume from. +func drainBatch(t *testing.T, batch *Batch) (values []string, next int64) { + t.Helper() + for { + msg, err := batch.ReadMessage() + if err != nil { + if !errors.Is(err, io.EOF) { + t.Fatalf("ReadMessage after %d messages: %v", len(values), err) + } + return values, batch.Offset() + } + values = append(values, string(msg.Value)) + } +} + +// TestBatchOffsetAdvancesPastTrailingControlBatch pins the offset bookkeeping +// for the layout a transactional producer actually writes: a commit marker +// closes the response. The marker is hidden from the caller, so nothing else +// records that its offset was consumed, and a reader that resumes from the +// marker fetches the same batch forever. +func TestBatchOffsetAdvancesPastTrailingControlBatch(t *testing.T) { + data := append( + v2Batch(t, batchOpts{firstOffset: 0, producerID: 7, transactional: true}, + v2Record(0, []byte("k0"), []byte("v0"))), + controlBatch(t, 1, 7, controlRecordCommit)..., + ) + + values, next := drainBatch(t, newTestBatch(t, 0, data)) + + if want := []string{"v0"}; !equalStrings(values, want) { + t.Errorf("values = %q, want %q", values, want) + } + if next != 2 { + t.Errorf("next offset = %d, want 2 (past the marker at offset 1)", next) + } +} + +// TestBatchOffsetAdvancesPastAllControlResponse is the second half of the same +// bug, and the more damaging one. Once the reader resumes at the marker, the +// next response holds nothing but that marker: no message is read, lastOffset +// keeps its zero value, and the offset fixup used to reset the partition to 1. +func TestBatchOffsetAdvancesPastAllControlResponse(t *testing.T) { + data := append( + controlBatch(t, 100, 7, controlRecordCommit), + controlBatch(t, 101, 8, controlRecordCommit)..., + ) + + values, next := drainBatch(t, newTestBatch(t, 100, data)) + + if len(values) != 0 { + t.Errorf("values = %q, want none", values) + } + if next != 102 { + t.Errorf("next offset = %d, want 102; a value of 1 is the partition rewind this guards", next) + } +} + +// TestBatchOffsetAdvancesPastAbortedTail covers the same accounting for a +// response whose tail is an aborted transaction rather than a lone marker. +func TestBatchOffsetAdvancesPastAbortedTail(t *testing.T) { + var data []byte + data = append(data, v2Batch(t, + batchOpts{firstOffset: 10, producerID: 7, transactional: true}, + v2Record(0, []byte("k0"), []byte("v0")))...) + data = append(data, controlBatch(t, 11, 7, controlRecordCommit)...) + data = append(data, v2Batch(t, + batchOpts{firstOffset: 12, producerID: 8, transactional: true}, + v2Record(0, []byte("k1"), []byte("aborted-0")), + v2Record(1, []byte("k2"), []byte("aborted-1")))...) + data = append(data, controlBatch(t, 14, 8, controlRecordAbort)...) + + values, next := drainBatch(t, newTestBatch(t, 10, data, + abortedTransaction{ProducerID: 8, FirstOffset: 12})) + + if want := []string{"v0"}; !equalStrings(values, want) { + t.Errorf("values = %q, want %q", values, want) + } + if next != 15 { + t.Errorf("next offset = %d, want 15 (past the abort marker at offset 14)", next) + } +} + +// TestBatchOffsetAdvancesPastTrailingEmptyControlBatch pins the offset +// bookkeeping for an emptied marker closing the response. Its offsets are +// consumed by the header read alone, and a reader that fails to account for +// them resumes at the empty batch and fetches it forever. +func TestBatchOffsetAdvancesPastTrailingEmptyControlBatch(t *testing.T) { + data := append( + v2Batch(t, batchOpts{firstOffset: 0, producerID: 7, transactional: true}, + v2Record(0, []byte("k0"), []byte("v0"))), + emptyBatch(t, batchOpts{firstOffset: 1, producerID: 7, control: true, transactional: true}, 0)..., + ) + + values, next := drainBatch(t, newTestBatch(t, 0, data)) + + if want := []string{"v0"}; !equalStrings(values, want) { + t.Errorf("values = %q, want %q", values, want) + } + if next != 2 { + t.Errorf("next offset = %d, want 2 (past the empty batch at offset 1)", next) + } +} + +// TestBatchOffsetAdvancesPastAllEmptyResponse is the refetch that follows once +// a reader resumes at an emptied batch: the response holds nothing but the +// empty batch, and the offset still has to move past it. +func TestBatchOffsetAdvancesPastAllEmptyResponse(t *testing.T) { + data := emptyBatch(t, batchOpts{firstOffset: 5, producerID: 7, control: true, transactional: true}, 0) + + values, next := drainBatch(t, newTestBatch(t, 5, data)) + + if len(values) != 0 { + t.Errorf("values = %q, want none", values) + } + if next != 6 { + t.Errorf("next offset = %d, want 6 (past the empty batch at offset 5)", next) + } +} + +// TestBatchOffsetNeverRewinds asserts the invariant the fixup relies on: a +// response that yields no message at all leaves the caller's position alone +// rather than moving it backwards. +func TestBatchOffsetNeverRewinds(t *testing.T) { + batch := newTestBatch(t, 500, controlBatch(t, 400, 7, controlRecordCommit)) + + if _, next := drainBatch(t, batch); next < 500 { + t.Errorf("next offset = %d, want no less than the 500 the batch started at", next) + } +} + +func equalStrings(a, b []string) bool { + if len(a) != len(b) { + return false + } + for i := range a { + if a[i] != b[i] { + return false + } + } + return true +} diff --git a/message_test.go b/message_test.go index 383cd226..3d6a51bc 100644 --- a/message_test.go +++ b/message_test.go @@ -715,10 +715,10 @@ type readerHelper struct { func newReaderHelper(t *testing.T, bs []byte) (r *readerHelper, err error) { bufReader := bufio.NewReader(bytes.NewReader(bs)) - _, _, remain, err := readFetchResponseHeaderV10(bufReader, len(bs)) + _, _, remain, aborted, err := readFetchResponseHeaderV10(bufReader, len(bs)) require.NoError(t, err) var msgs *messageSetReader - msgs, err = newMessageSetReader(bufReader, remain) + msgs, err = newMessageSetReader(bufReader, remain, aborted) if err != nil { return } diff --git a/read.go b/read.go index ec2b3852..b6c3afa3 100644 --- a/read.go +++ b/read.go @@ -304,7 +304,20 @@ func readSlice(r *bufio.Reader, sz int, v reflect.Value) (int, error) { return sz, nil } -func readFetchResponseHeaderV2(r *bufio.Reader, size int) (throttle int32, watermark int64, remain int, err error) { +// abortedTransaction is one entry of the aborted transaction list a broker +// returns in a fetch response. The broker only populates the list when the +// request asked for ReadCommitted; it bounds the response at the last stable +// offset and leaves it to the client to drop the records these entries point +// at. +// +// The fields have to stay exported and in this order: readStruct reads them +// positionally through reflection, which cannot address unexported fields. +type abortedTransaction struct { + ProducerID int64 + FirstOffset int64 +} + +func readFetchResponseHeaderV2(r *bufio.Reader, size int) (throttle int32, watermark int64, remain int, aborted []abortedTransaction, err error) { var n int32 var p struct { Partition int32 @@ -366,12 +379,8 @@ func readFetchResponseHeaderV2(r *bufio.Reader, size int) (throttle int32, water return } -func readFetchResponseHeaderV5(r *bufio.Reader, size int) (throttle int32, watermark int64, remain int, err error) { +func readFetchResponseHeaderV5(r *bufio.Reader, size int) (throttle int32, watermark int64, remain int, aborted []abortedTransaction, err error) { var n int32 - type AbortedTransaction struct { - ProducerId int64 - FirstOffset int64 - } var p struct { Partition int32 ErrorCode int16 @@ -380,7 +389,6 @@ func readFetchResponseHeaderV5(r *bufio.Reader, size int) (throttle int32, water LogStartOffset int64 } var messageSetSize int32 - var abortedTransactions []AbortedTransaction if remain, err = readInt32(r, size, &throttle); err != nil { return @@ -424,12 +432,12 @@ func readFetchResponseHeaderV5(r *bufio.Reader, size int) (throttle int32, water return } - if abortedTransactionLen == -1 { - abortedTransactions = nil - } else { - abortedTransactions = make([]AbortedTransaction, abortedTransactionLen) + // A length of -1 is the protocol's null array, and a length of 0 is an + // empty one; neither leaves anything for the reader to filter. + if abortedTransactionLen > 0 { + aborted = make([]abortedTransaction, abortedTransactionLen) for i := 0; i < abortedTransactionLen; i++ { - if remain, err = read(r, remain, &abortedTransactions[i]); err != nil { + if remain, err = read(r, remain, &aborted[i]); err != nil { return } } @@ -457,13 +465,9 @@ func readFetchResponseHeaderV5(r *bufio.Reader, size int) (throttle int32, water } -func readFetchResponseHeaderV10(r *bufio.Reader, size int) (throttle int32, watermark int64, remain int, err error) { +func readFetchResponseHeaderV10(r *bufio.Reader, size int) (throttle int32, watermark int64, remain int, aborted []abortedTransaction, err error) { var n int32 var errorCode int16 - type AbortedTransaction struct { - ProducerId int64 - FirstOffset int64 - } var p struct { Partition int32 ErrorCode int16 @@ -472,7 +476,6 @@ func readFetchResponseHeaderV10(r *bufio.Reader, size int) (throttle int32, wate LogStartOffset int64 } var messageSetSize int32 - var abortedTransactions []AbortedTransaction if remain, err = readInt32(r, size, &throttle); err != nil { return @@ -528,12 +531,12 @@ func readFetchResponseHeaderV10(r *bufio.Reader, size int) (throttle int32, wate return } - if abortedTransactionLen == -1 { - abortedTransactions = nil - } else { - abortedTransactions = make([]AbortedTransaction, abortedTransactionLen) + // A length of -1 is the protocol's null array, and a length of 0 is an + // empty one; neither leaves anything for the reader to filter. + if abortedTransactionLen > 0 { + aborted = make([]abortedTransaction, abortedTransactionLen) for i := 0; i < abortedTransactionLen; i++ { - if remain, err = read(r, remain, &abortedTransactions[i]); err != nil { + if remain, err = read(r, remain, &aborted[i]); err != nil { return } } diff --git a/reader.go b/reader.go index 04d90f35..ed874a06 100644 --- a/reader.go +++ b/reader.go @@ -510,6 +510,9 @@ type ReaderConfig struct { // IsolationLevel controls the visibility of transactional records. // ReadUncommitted makes all records visible. With ReadCommitted only // non-transactional and committed records are visible. + // + // Defaults to ReadUncommitted, matching Kafka's own default. See the + // IsolationLevel constants for the trade-off ReadCommitted carries. IsolationLevel IsolationLevel // Limit of how many attempts to connect will be made before returning the error. @@ -1534,6 +1537,15 @@ func (r *reader) read(ctx context.Context, offset int64, conn *Conn) (int64, err bytes += n } + // The tail of a response can be control batches or batches from aborted + // transactions, which are consumed without producing a message. The batch + // tracked those offsets, so take its position when it is ahead of the last + // message read; otherwise a reconnect would resume before them and redeliver + // everything in between. + if o := batch.Offset(); o > offset { + offset = o + } + conn.SetReadDeadline(time.Time{}) t2 := time.Now()