diff --git a/batch.go b/batch.go index eb742712d..0902545b9 100644 --- a/batch.go +++ b/batch.go @@ -28,7 +28,11 @@ type Batch struct { partition int offset int64 highWaterMark int64 - err error + // preferredReadReplica is the broker id (>=0) the broker would prefer + // the consumer fetch from for this partition (KIP-392). -1 means no + // preference (fetch from leader). + preferredReadReplica int32 + err error // The last offset in the batch. // // We use lastOffset to skip offsets that have been compacted away. @@ -51,6 +55,17 @@ func (batch *Batch) HighWaterMark() int64 { return batch.highWaterMark } +// PreferredReadReplica returns the broker id of the replica the kafka broker +// recommends the consumer fetch from for this partition, as introduced in +// KIP-392 (Allow Consumers to Fetch from Closest Replica). A value of -1 +// means no preference and the consumer should keep fetching from the +// partition leader. The value is only populated when the connection +// negotiated Fetch v11 or higher and the broker chose a preferred replica; +// otherwise it is -1. +func (batch *Batch) PreferredReadReplica() int32 { + return batch.preferredReadReplica +} + // Partition returns the batch partition. func (batch *Batch) Partition() int { return batch.partition diff --git a/conn.go b/conn.go index 9f9f25903..24c988899 100644 --- a/conn.go +++ b/conn.go @@ -54,6 +54,7 @@ type Conn struct { fetchMinSize int32 broker int32 rack string + clientRack string // correlation ID generator (synchronized on wlock) correlationID int32 @@ -91,6 +92,13 @@ type ConnConfig struct { Broker int Rack string + // ClientRack is the consumer's rack id used by the broker to determine + // the closest replica to fetch from (KIP-392). When set and the broker + // supports Fetch v11 or higher, the rack id is sent in fetch requests + // and the broker may direct the consumer to read from a follower + // replica that is closer to the consumer. + ClientRack string + // The transactional id to use for transactional delivery. Idempotent // deliver should be enabled if transactional id is configured. // For more details look at transactional.id description here: http://kafka.apache.org/documentation.html#producerconfigs @@ -179,6 +187,7 @@ func NewConnWith(conn net.Conn, config ConnConfig) *Conn { partition: int32(config.Partition), broker: int32(config.Broker), rack: config.Rack, + clientRack: config.ClientRack, offset: FirstOffset, requiredAcks: -1, transactionalID: emptyToNullable(config.TransactionalID), @@ -777,7 +786,7 @@ func (c *Conn) ReadBatchWith(cfg ReadBatchConfig) *Batch { return &Batch{err: dontExpectEOF(err)} } - fetchVersion, err := c.negotiateVersion(fetch, v2, v5, v10) + fetchVersion, err := c.negotiateVersion(fetch, v2, v5, v10, v11) if err != nil { return &Batch{err: dontExpectEOF(err)} } @@ -799,6 +808,19 @@ func (c *Conn) ReadBatchWith(cfg ReadBatchConfig) *Batch { // truncated messages. adjustedDeadline = deadline switch fetchVersion { + case v11: + return c.wb.writeFetchRequestV11( + id, + c.clientID, + c.topic, + c.partition, + offset, + cfg.MinBytes, + cfg.MaxBytes+int(c.fetchMinSize), + timeout, + int8(cfg.IsolationLevel), + c.clientRack, + ) case v10: return c.wb.writeFetchRequestV10( id, @@ -848,8 +870,11 @@ func (c *Conn) ReadBatchWith(cfg ReadBatchConfig) *Batch { var throttle int32 var highWaterMark int64 var remain int + var preferredReadReplica int32 = -1 switch fetchVersion { + case v11: + throttle, highWaterMark, preferredReadReplica, remain, err = readFetchResponseHeaderV11(&c.rbuf, size) case v10: throttle, highWaterMark, remain, err = readFetchResponseHeaderV10(&c.rbuf, size) case v5: @@ -874,15 +899,16 @@ func (c *Conn) ReadBatchWith(cfg ReadBatchConfig) *Batch { } return &Batch{ - conn: c, - msgs: msgs, - deadline: adjustedDeadline, - throttle: makeDuration(throttle), - lock: lock, - topic: c.topic, // topic is copied to Batch to prevent race with Batch.close - partition: int(c.partition), // partition is copied to Batch to prevent race with Batch.close - offset: offset, - highWaterMark: highWaterMark, + conn: c, + msgs: msgs, + deadline: adjustedDeadline, + throttle: makeDuration(throttle), + lock: lock, + topic: c.topic, // topic is copied to Batch to prevent race with Batch.close + partition: int(c.partition), // partition is copied to Batch to prevent race with Batch.close + offset: offset, + highWaterMark: highWaterMark, + preferredReadReplica: preferredReadReplica, // there shouldn't be a short read on initially setting up the batch. // as such, any io.EOF is re-mapped to an io.ErrUnexpectedEOF so that we // don't accidentally signal that we successfully reached the end of the diff --git a/dialer.go b/dialer.go index 7786ed320..18a506173 100644 --- a/dialer.go +++ b/dialer.go @@ -89,6 +89,11 @@ type Dialer struct { // For more details look at transactional.id description here: http://kafka.apache.org/documentation.html#producerconfigs // Empty string means that the connection will be non-transactional. TransactionalID string + + // ClientRack is the consumer's rack id (KIP-392). When set and the broker + // supports Fetch v11+, the broker may direct the consumer to fetch from + // the closest replica instead of the partition leader. + ClientRack string } // Dial connects to the address on the named network. @@ -117,6 +122,7 @@ func (d *Dialer) DialContext(ctx context.Context, network string, address string ConnConfig{ ClientID: d.ClientID, TransactionalID: d.TransactionalID, + ClientRack: d.ClientRack, }, ) } @@ -148,6 +154,7 @@ func (d *Dialer) DialPartition(ctx context.Context, network string, address stri Broker: partition.Leader.ID, Rack: partition.Leader.Rack, TransactionalID: d.TransactionalID, + ClientRack: d.ClientRack, }) } diff --git a/fetch.go b/fetch.go index eafd0de88..71e332a24 100644 --- a/fetch.go +++ b/fetch.go @@ -37,6 +37,14 @@ type FetchRequest struct { // This field requires the kafka broker to support the Fetch API in version // 4 or above (otherwise the value is ignored). IsolationLevel IsolationLevel + + // RackID is the consumer's rack id (KIP-392). When set, the broker may + // direct the consumer to fetch from the closest replica via the + // PreferredReadReplica field of the response. + // + // This field requires the kafka broker to support the Fetch API in + // version 11 or above (otherwise the value is ignored). + RackID string } // FetchResponse represents a response from a kafka broker to a fetch request. @@ -60,6 +68,13 @@ type FetchResponse struct { LastStableOffset int64 LogStartOffset int64 + // PreferredReadReplica is the broker id the broker would prefer the + // consumer fetch from for this partition (KIP-392). A value of -1 means + // no preference. Only populated when the broker supports Fetch v11 or + // above AND the caller set RackID on the request; in any other case + // the value is forced to -1. + PreferredReadReplica int32 + // An error that may have occurred while attempting to fetch the records. // // The error contains both the kafka error code, and an error message @@ -145,6 +160,7 @@ func (c *Client) Fetch(ctx context.Context, req *FetchRequest) (*FetchResponse, PartitionMaxBytes: int32(req.MaxBytes), }}, }}, + RackID: req.RackID, }) if err != nil { @@ -162,14 +178,26 @@ func (c *Client) Fetch(ctx context.Context, req *FetchRequest) (*FetchResponse, partition := &topic.Partitions[0] ret := &FetchResponse{ - Throttle: makeDuration(res.ThrottleTimeMs), - Topic: topic.Topic, - Partition: int(partition.Partition), - Error: makeError(res.ErrorCode, ""), - HighWatermark: partition.HighWatermark, - LastStableOffset: partition.LastStableOffset, - LogStartOffset: partition.LogStartOffset, - Records: partition.RecordSet.Records, + Throttle: makeDuration(res.ThrottleTimeMs), + Topic: topic.Topic, + Partition: int(partition.Partition), + Error: makeError(res.ErrorCode, ""), + HighWatermark: partition.HighWatermark, + LastStableOffset: partition.LastStableOffset, + LogStartOffset: partition.LogStartOffset, + PreferredReadReplica: partition.PreferredReadReplica, + Records: partition.RecordSet.Records, + } + + // KIP-392: PreferredReadReplica is only meaningful in Fetch v11+ responses. + // When the broker negotiated an older version, the protocol decoder leaves + // the field at Go's zero value (0), which is indistinguishable from broker + // id 0 being the preferred replica. We can't read the negotiated version + // from here, but we know that if the caller did not set RackID then they + // did not opt into KIP-392 and we should not surface any preferred replica + // id at all. Force -1 so callers can rely on "-1 means no preference". + if req.RackID == "" { + ret.PreferredReadReplica = -1 } if partition.ErrorCode != 0 { diff --git a/fetch_kip392_test.go b/fetch_kip392_test.go new file mode 100644 index 000000000..95952155d --- /dev/null +++ b/fetch_kip392_test.go @@ -0,0 +1,107 @@ +package kafka + +import ( + "context" + "net" + "testing" + "time" + + "github.com/segmentio/kafka-go/protocol" + fetchAPI "github.com/segmentio/kafka-go/protocol/fetch" +) + +// stubRoundTripper returns a canned response for any request it receives. +type stubRoundTripper struct { + resp protocol.Message + err error +} + +func (s *stubRoundTripper) RoundTrip(_ context.Context, _ net.Addr, _ protocol.Message) (protocol.Message, error) { + return s.resp, s.err +} + +// TestClientFetch_PreferredReadReplicaDefaultsMinusOneWithoutRackID asserts +// that Client.Fetch surfaces PreferredReadReplica = -1 when the caller did +// not opt into KIP-392 (req.RackID == ""). Without this guard, a v10 broker +// response (or any response that omits the v11 field) would surface the Go +// zero value 0, indistinguishable from broker id 0 being preferred. +func TestClientFetch_PreferredReadReplicaDefaultsMinusOneWithoutRackID(t *testing.T) { + stub := &stubRoundTripper{ + resp: &fetchAPI.Response{ + ThrottleTimeMs: 0, + Topics: []fetchAPI.ResponseTopic{{ + Topic: "topic-x", + Partitions: []fetchAPI.ResponsePartition{{ + Partition: 0, + ErrorCode: 0, + HighWatermark: 100, + LastStableOffset: 100, + LogStartOffset: 0, + PreferredReadReplica: 0, // simulates v10 zero-value field + }}, + }}, + }, + } + + client := &Client{ + Addr: TCP("127.0.0.1:9092"), + Timeout: time.Second, + Transport: stub, + } + + resp, err := client.Fetch(context.Background(), &FetchRequest{ + Topic: "topic-x", + Partition: 0, + Offset: 0, + MinBytes: 1, + MaxBytes: 1024, + MaxWait: 100 * time.Millisecond, + // RackID intentionally left empty -- caller did not opt in. + }) + if err != nil { + t.Fatalf("Client.Fetch: %v", err) + } + if resp.PreferredReadReplica != -1 { + t.Fatalf("PreferredReadReplica: got %d, want -1 when RackID is empty", resp.PreferredReadReplica) + } +} + +// TestClientFetch_PreferredReadReplicaPassThroughWithRackID asserts that when +// the caller did opt into KIP-392 by setting RackID, broker id 0 is passed +// through verbatim (broker 0 may legitimately be the preferred replica). +func TestClientFetch_PreferredReadReplicaPassThroughWithRackID(t *testing.T) { + stub := &stubRoundTripper{ + resp: &fetchAPI.Response{ + Topics: []fetchAPI.ResponseTopic{{ + Topic: "topic-x", + Partitions: []fetchAPI.ResponsePartition{{ + Partition: 0, + HighWatermark: 100, + PreferredReadReplica: 7, + }}, + }}, + }, + } + + client := &Client{ + Addr: TCP("127.0.0.1:9092"), + Timeout: time.Second, + Transport: stub, + } + + resp, err := client.Fetch(context.Background(), &FetchRequest{ + Topic: "topic-x", + Partition: 0, + Offset: 0, + MinBytes: 1, + MaxBytes: 1024, + MaxWait: 100 * time.Millisecond, + RackID: "rack-nl", + }) + if err != nil { + t.Fatalf("Client.Fetch: %v", err) + } + if resp.PreferredReadReplica != 7 { + t.Fatalf("PreferredReadReplica: got %d, want 7 (passed through verbatim)", resp.PreferredReadReplica) + } +} diff --git a/fetch_v11_wire_test.go b/fetch_v11_wire_test.go new file mode 100644 index 000000000..0924073b0 --- /dev/null +++ b/fetch_v11_wire_test.go @@ -0,0 +1,138 @@ +package kafka + +import ( + "bufio" + "bytes" + "strings" + "testing" + "time" +) + +// TestWriteFetchRequestV11_ContainsRackID asserts that writeFetchRequestV11 +// actually places the rack id into the on-wire request body, and that the v10 +// writer does not (so we don't accidentally regress the older path). +func TestWriteFetchRequestV11_ContainsRackID(t *testing.T) { + const ( + correlationID = 1 + clientID = "client" + topic = "topic-x" + partition = int32(0) + offset = int64(0) + minBytes = 1 + maxBytes = 1024 + maxWait = 100 * time.Millisecond + isoLevel = int8(0) + rack = "rack-nl" + ) + + v11Buf := &bytes.Buffer{} + v11WB := &writeBuffer{w: v11Buf} + if err := v11WB.writeFetchRequestV11(correlationID, clientID, topic, partition, offset, minBytes, maxBytes, maxWait, isoLevel, rack); err != nil { + t.Fatalf("writeFetchRequestV11: %v", err) + } + if !bytes.Contains(v11Buf.Bytes(), []byte(rack)) { + t.Fatalf("v11 request bytes do not contain rack id %q", rack) + } + + v10Buf := &bytes.Buffer{} + v10WB := &writeBuffer{w: v10Buf} + if err := v10WB.writeFetchRequestV10(correlationID, clientID, topic, partition, offset, minBytes, maxBytes, maxWait, isoLevel); err != nil { + t.Fatalf("writeFetchRequestV10: %v", err) + } + if bytes.Contains(v10Buf.Bytes(), []byte(rack)) { + t.Fatalf("v10 request bytes unexpectedly contain rack id %q", rack) + } +} + +// TestReadFetchResponseHeaderV11_DecodesPreferredReadReplica builds a minimal +// v11 fetch response by hand and verifies that the decoder recovers the +// preferred read replica id and defaults to -1 when the broker reports no +// preference. +func TestReadFetchResponseHeaderV11_DecodesPreferredReadReplica(t *testing.T) { + cases := []struct { + name string + prr int32 + }{ + {"explicit-preferred-replica", 7}, + {"no-preference-minus-one", -1}, + // broker id 0 is a valid id; make sure the decoder returns it + // verbatim and doesn't conflate it with "no preference". + {"broker-id-zero", 0}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + buf := buildFetchResponseV11(t, tc.prr) + r := bufio.NewReader(bytes.NewReader(buf)) + throttle, watermark, prr, remain, err := readFetchResponseHeaderV11(r, len(buf)) + if err != nil { + t.Fatalf("readFetchResponseHeaderV11: %v", err) + } + if throttle != 0 { + t.Errorf("throttle: got %d, want 0", throttle) + } + if watermark != 1000 { + t.Errorf("watermark: got %d, want 1000", watermark) + } + if prr != tc.prr { + t.Errorf("preferredReadReplica: got %d, want %d", prr, tc.prr) + } + if remain != 0 { + t.Errorf("remain: got %d, want 0 (no record set bytes left)", remain) + } + }) + } +} + +// buildFetchResponseV11 hand-assembles a v11 fetch response body containing one +// topic and one partition with the given preferred read replica and an empty +// record set. The shape mirrors what the broker would put on the wire so the +// decoder gets exercised end-to-end. +func buildFetchResponseV11(t *testing.T, preferredReadReplica int32) []byte { + t.Helper() + buf := &bytes.Buffer{} + wb := &writeBuffer{w: buf} + + wb.writeInt32(0) // throttle time ms + wb.writeInt16(0) // top-level error code (v7+) + wb.writeInt32(-1) // session id (v7+) + + // topics array + wb.writeArrayLen(1) + wb.writeString("topic-x") + + // partitions array + wb.writeArrayLen(1) + wb.writeInt32(0) // partition index + wb.writeInt16(0) // error code + wb.writeInt64(1000) // high watermark + wb.writeInt64(900) // last stable offset + wb.writeInt64(0) // log start offset + wb.writeArrayLen(0) // aborted transactions = empty + + // preferred read replica (KIP-392) + wb.writeInt32(preferredReadReplica) + + // message set size = 0 (empty record set) + wb.writeInt32(0) + + return buf.Bytes() +} + +// TestReadFetchResponseHeaderV11_NoPreferenceDefault asserts the decoder +// initializes preferredReadReplica to -1 before reading anything from the +// wire, so a malformed/short response surfaces a sensible default rather +// than an accidental "broker 0 is preferred" value. +func TestReadFetchResponseHeaderV11_NoPreferenceDefault(t *testing.T) { + // A truncated response (just the throttle bytes) should fail to decode + // but the returned prr should still be -1, not 0. + buf := []byte{0, 0, 0, 0} // 4 bytes of throttle, then EOF + r := bufio.NewReader(bytes.NewReader(buf)) + _, _, prr, _, err := readFetchResponseHeaderV11(r, len(buf)) + if err == nil { + t.Fatalf("expected an error decoding truncated response, got nil") + } + if !strings.Contains(err.Error(), "") || prr != -1 { + t.Fatalf("preferredReadReplica default: got %d, want -1", prr) + } +} diff --git a/protocol.go b/protocol.go index 37208abf1..ca7191766 100644 --- a/protocol.go +++ b/protocol.go @@ -110,6 +110,7 @@ const ( v6 = 6 v7 = 7 v10 = 10 + v11 = 11 // Unused protocol versions: v4, v8, v9. ) diff --git a/protocol/fetch/fetch_test.go b/protocol/fetch/fetch_test.go index 3f7f820d9..059738d74 100644 --- a/protocol/fetch/fetch_test.go +++ b/protocol/fetch/fetch_test.go @@ -32,6 +32,32 @@ func TestFetchRequest(t *testing.T) { }, }, }) + + // KIP-392: round-trip a v11 request with RackID set to make sure the + // rack id is encoded and decoded unchanged. + prototest.TestRequest(t, v11, &fetch.Request{ + ReplicaID: -1, + MaxWaitTime: 500, + MinBytes: 1024, + MaxBytes: 1 << 20, + SessionID: -1, + SessionEpoch: -1, + Topics: []fetch.RequestTopic{ + { + Topic: "topic-1", + Partitions: []fetch.RequestPartition{ + { + Partition: 1, + CurrentLeaderEpoch: -1, + FetchOffset: 2, + LogStartOffset: -1, + PartitionMaxBytes: 1024, + }, + }, + }, + }, + RackID: "rack-nl", + }) } func TestFetchResponse(t *testing.T) { @@ -73,8 +99,9 @@ func TestFetchResponse(t *testing.T) { Topic: "topic-1", Partitions: []fetch.ResponsePartition{ { - Partition: 1, - HighWatermark: 1000, + Partition: 1, + HighWatermark: 1000, + PreferredReadReplica: 42, RecordSet: protocol.RecordSet{ Version: 2, Records: protocol.NewRecordReader( @@ -88,6 +115,30 @@ func TestFetchResponse(t *testing.T) { }, }, }) + + // KIP-392: when no preferred replica exists the broker sends -1. + // Make sure that value round-trips so the client doesn't accidentally + // flip it to 0 (which is a valid broker id). + prototest.TestResponse(t, v11, &fetch.Response{ + Topics: []fetch.ResponseTopic{ + { + Topic: "topic-1", + Partitions: []fetch.ResponsePartition{ + { + Partition: 1, + HighWatermark: 1000, + PreferredReadReplica: -1, + RecordSet: protocol.RecordSet{ + Version: 2, + Records: protocol.NewRecordReader( + protocol.Record{Offset: 0, Time: t0, Key: nil, Value: prototest.String("msg-0")}, + ), + }, + }, + }, + }, + }, + }) } func BenchmarkFetchResponse(b *testing.B) { diff --git a/read.go b/read.go index ec2b38527..a3c8a7931 100644 --- a/read.go +++ b/read.go @@ -560,3 +560,107 @@ func readFetchResponseHeaderV10(r *bufio.Reader, size int) (throttle int32, wate return } + +// readFetchResponseHeaderV11 mirrors readFetchResponseHeaderV10 but additionally +// decodes the per-partition PreferredReadReplica field introduced by KIP-392 +// (Fetch v11). Returned preferredReadReplica is -1 when the broker has no +// preference. +func readFetchResponseHeaderV11(r *bufio.Reader, size int) (throttle int32, watermark int64, preferredReadReplica int32, remain int, err error) { + var n int32 + var errorCode int16 + type AbortedTransaction struct { + ProducerId int64 + FirstOffset int64 + } + var p struct { + Partition int32 + ErrorCode int16 + HighwaterMarkOffset int64 + LastStableOffset int64 + LogStartOffset int64 + } + var messageSetSize int32 + var abortedTransactions []AbortedTransaction + preferredReadReplica = -1 + + if remain, err = readInt32(r, size, &throttle); err != nil { + return + } + + if remain, err = readInt16(r, remain, &errorCode); err != nil { + return + } + if errorCode != 0 { + err = Error(errorCode) + return + } + + if remain, err = discardInt32(r, remain); err != nil { + return + } + + if remain, err = readInt32(r, remain, &n); err != nil { + return + } + if n != 1 { + err = fmt.Errorf("1 kafka topic was expected in the fetch response but the client received %d", n) + return + } + + if remain, err = discardString(r, remain); err != nil { + return + } + + if remain, err = readInt32(r, remain, &n); err != nil { + return + } + if n != 1 { + err = fmt.Errorf("1 kafka partition was expected in the fetch response but the client received %d", n) + return + } + + if remain, err = read(r, remain, &p); err != nil { + return + } + + var abortedTransactionLen int + if remain, err = readArrayLen(r, remain, &abortedTransactionLen); err != nil { + return + } + + if abortedTransactionLen == -1 { + abortedTransactions = nil + } else { + abortedTransactions = make([]AbortedTransaction, abortedTransactionLen) + for i := 0; i < abortedTransactionLen; i++ { + if remain, err = read(r, remain, &abortedTransactions[i]); err != nil { + return + } + } + } + _ = abortedTransactions + + // PreferredReadReplica (KIP-392): broker id of the preferred read replica + // for this partition, or -1 if no preference. + if remain, err = readInt32(r, remain, &preferredReadReplica); err != nil { + return + } + + if p.ErrorCode != 0 { + err = Error(p.ErrorCode) + return + } + + remain, err = readInt32(r, remain, &messageSetSize) + if err != nil { + return + } + + if remain != int(messageSetSize) { + err = fmt.Errorf("the size of the message set in a fetch response doesn't match the number of remaining bytes (message set size = %d, remaining bytes = %d)", messageSetSize, remain) + return + } + + watermark = p.HighwaterMarkOffset + return +} diff --git a/reader.go b/reader.go index 04d90f355..a6f366a1b 100644 --- a/reader.go +++ b/reader.go @@ -522,6 +522,13 @@ type ReaderConfig struct { // This flag is being added to retain backwards-compatibility, so it will be // removed in a future version of kafka-go. OffsetOutOfRangeError bool + + // Rack is the consumer's rack id (KIP-392 client.rack). When set, the + // reader advertises this rack to the broker on each fetch request and + // the broker may direct the consumer to fetch from the closest replica + // instead of the partition leader. Requires the broker to support + // Fetch v11 or higher; ignored otherwise. + Rack string } // Validate method validates ReaderConfig properties. @@ -644,6 +651,15 @@ func NewReader(config ReaderConfig) *Reader { config.Dialer = DefaultDialer } + // Propagate the consumer rack (KIP-392) onto the dialer so that fetch + // requests sent through Conns dialed by the reader carry the rack id. + // Clone the dialer to avoid mutating a shared instance. + if config.Rack != "" && config.Dialer.ClientRack != config.Rack { + d := *config.Dialer + d.ClientRack = config.Rack + config.Dialer = &d + } + if config.MaxBytes == 0 { config.MaxBytes = 1e6 // 1 MB } @@ -1196,23 +1212,25 @@ func (r *Reader) start(offsetsByPartition map[topicPartition]int64) { defer join.Done() (&reader{ - dialer: r.config.Dialer, - logger: r.config.Logger, - errorLogger: r.config.ErrorLogger, - brokers: r.config.Brokers, - topic: key.topic, - partition: int(key.partition), - minBytes: r.config.MinBytes, - maxBytes: r.config.MaxBytes, - maxWait: r.config.MaxWait, - readBatchTimeout: r.config.ReadBatchTimeout, - backoffDelayMin: r.config.ReadBackoffMin, - backoffDelayMax: r.config.ReadBackoffMax, - version: r.version, - msgs: r.msgs, - stats: r.stats, - isolationLevel: r.config.IsolationLevel, - maxAttempts: r.config.MaxAttempts, + dialer: r.config.Dialer, + logger: r.config.Logger, + errorLogger: r.config.ErrorLogger, + brokers: r.config.Brokers, + topic: key.topic, + partition: int(key.partition), + minBytes: r.config.MinBytes, + maxBytes: r.config.MaxBytes, + maxWait: r.config.MaxWait, + readBatchTimeout: r.config.ReadBatchTimeout, + backoffDelayMin: r.config.ReadBackoffMin, + backoffDelayMax: r.config.ReadBackoffMax, + version: r.version, + msgs: r.msgs, + stats: r.stats, + isolationLevel: r.config.IsolationLevel, + maxAttempts: r.config.MaxAttempts, + rack: r.config.Rack, + preferredReadReplica: -1, // backwards-compatibility flags offsetOutOfRangeError: r.config.OffsetOutOfRangeError, @@ -1243,9 +1261,73 @@ type reader struct { isolationLevel IsolationLevel maxAttempts int + // rack is the consumer's rack id (KIP-392). Empty means disabled. + rack string + // preferredReadReplica is the broker id the broker last asked us to + // fetch from for this partition (-1 = no preference / fetch from leader). + preferredReadReplica int32 + // preferredReadReplicaExpiresAt bounds the time we keep using the + // preferred replica before falling back to the leader and re-discovering. + preferredReadReplicaExpiresAt time.Time + // preferredReadReplicaCooldown is a per-replica negative cache. When a + // preferred replica fails (cannot dial, NotLeader, OffsetOutOfRange, + // etc.) we record an "ignore until" timestamp here. Subsequent leader + // Fetch responses that re-advertise the same broker id are ignored + // until the cooldown expires. Without this the broker would re-suggest + // the same bad replica on every Fetch and the reader would live-lock, + // because the broker's metadata is per-cluster and does not learn from + // our local failure. + preferredReadReplicaCooldown map[int32]time.Time + + // connectedToFollower is set by initialize() when the *Conn it just + // returned is talking to a KIP-392 preferred follower (as opposed to + // the partition leader). read() uses it to disambiguate two semantically + // different "PreferredReadReplica = -1" cases in a FetchResponse: + // - From the leader: the broker has revoked the follower preference + // for this client; we should genuinely fall back. + // - From the follower we are already pinned to: the follower has no + // further redirect to suggest (it does not advertise itself as its + // own preference). This is the normal steady-state response and + // must NOT be interpreted as revocation, otherwise we would + // reconnect to the leader on the very first follower fetch and + // immediately get re-redirected back, oscillating until the + // broker's selector eventually stops re-asserting. + connectedToFollower bool + offsetOutOfRangeError bool } +// preferredReplicaInCooldown returns true if id is currently suppressed by +// the per-reader negative cache. Expired entries are pruned. +func (r *reader) preferredReplicaInCooldown(id int32) bool { + if r.preferredReadReplicaCooldown == nil { + return false + } + until, ok := r.preferredReadReplicaCooldown[id] + if !ok { + return false + } + if time.Now().After(until) { + delete(r.preferredReadReplicaCooldown, id) + return false + } + return true +} + +// blockPreferredReplica suppresses id for the next 5 minutes, matching the +// preferred-replica TTL used elsewhere. Called whenever a follower-fetch +// attempt against id has failed in a way that makes it pointless to retry +// immediately. +func (r *reader) blockPreferredReplica(id int32) { + if id < 0 { + return + } + if r.preferredReadReplicaCooldown == nil { + r.preferredReadReplicaCooldown = make(map[int32]time.Time) + } + r.preferredReadReplicaCooldown[id] = time.Now().Add(5 * time.Minute) +} + type readerMessage struct { version int64 message Message @@ -1346,6 +1428,13 @@ func (r *reader) run(ctx context.Context, offset int64) { conn.Close() break readLoop + case errors.Is(err, errPreferredReadReplicaChanged): + // KIP-392: the broker selected a (new) preferred read replica + // for us, or our preference expired. Reconnect via the outer + // loop using the updated preferredReadReplica state. + errcount = 0 + break readLoop + case errors.Is(err, UnknownTopicOrPartition): r.withErrorLogger(func(log Logger) { log.Printf("failed to read from current broker %v for partition %d of %s at offset %d: %v", r.brokers, r.partition, r.topic, toHumanOffset(offset), err) @@ -1363,6 +1452,19 @@ func (r *reader) run(ctx context.Context, offset int64) { log.Printf("failed to read from current broker for partition %d of %s at offset %d: %v", r.partition, r.topic, toHumanOffset(offset), err) }) + // KIP-392: if the error came while connected to a preferred + // follower, clear the preference so the next initialize + // dials the actual leader instead of looping back to the + // same follower. + if r.preferredReadReplica >= 0 { + r.withLogger(func(log Logger) { + log.Printf("kafka reader clearing preferred replica %d for partition %d of %s after NotLeaderForPartition", r.preferredReadReplica, r.partition, r.topic) + }) + r.blockPreferredReplica(r.preferredReadReplica) + r.preferredReadReplica = -1 + r.preferredReadReplicaExpiresAt = time.Time{} + } + conn.Close() // The next call to .initialize will re-establish a connection to the proper @@ -1380,6 +1482,25 @@ func (r *reader) run(ctx context.Context, offset int64) { continue case errors.Is(err, OffsetOutOfRange): + // KIP-392: an OffsetOutOfRange from a follower is not + // authoritative because follower offsets can lag behind + // the leader's. Drop the preference, close the + // connection, and let the run loop reinitialize through + // the leader on the next iteration. The leader's view of + // first/last is the only one we trust for the reset + // decision. + if r.preferredReadReplica >= 0 { + r.withLogger(func(log Logger) { + log.Printf("kafka reader got OffsetOutOfRange from preferred replica %d for partition %d of %s, falling back to leader before deciding", r.preferredReadReplica, r.partition, r.topic) + }) + r.blockPreferredReplica(r.preferredReadReplica) + r.preferredReadReplica = -1 + r.preferredReadReplicaExpiresAt = time.Time{} + conn.Close() + errcount = 0 + break readLoop + } + first, last, err := r.readOffsets(conn) if err != nil { r.withErrorLogger(func(log Logger) { @@ -1440,24 +1561,132 @@ func (r *reader) run(ctx context.Context, offset int64) { } func (r *reader) initialize(ctx context.Context, offset int64) (conn *Conn, start int64, err error) { + // Default to "not on a follower" until a follower-dial succeeds below. + // Reset on every initialize() so that a previous follower session that + // got torn down doesn't leak state into a leader-only attempt. + r.connectedToFollower = false + + // If a preferred read replica was selected (KIP-392) and is still fresh, + // try to dial it. On failure (broker unknown, dial error, etc.) we fall + // back to the leader. + usePreferred := r.preferredReadReplica >= 0 && time.Now().Before(r.preferredReadReplicaExpiresAt) + for i := 0; i != len(r.brokers) && conn == nil; i++ { broker := r.brokers[i] var first, last int64 + // onFollower tracks whether the connection we're about to seek on is + // a KIP-392 preferred follower. Followers reject ListOffsets with + // NotLeaderForPartition, so we must skip the validating ListOffsets + // call inside conn.Seek (SeekDontCheck) when seeking on a follower. + onFollower := false + + // KIP-392 only specifies that *Fetch* may be served by a follower; + // ListOffsets is still leader-only and a follower will reject it + // with NotLeaderForPartition. So if we plan to fetch from a + // preferred follower: + // - For a concrete offset we already know what to seek to and + // can skip readOffsets entirely. + // - For the FirstOffset / LastOffset markers we MUST resolve + // them against the leader first, then dial the follower. + needOffsetResolution := offset == FirstOffset || offset == LastOffset + + if usePreferred && !needOffsetResolution { + t0 := time.Now() + conn, err = r.dialPreferredReplica(ctx, broker) + t1 := time.Now() + r.stats.dials.observe(1) + r.stats.dialTime.observeDuration(t1.Sub(t0)) + + if err != nil { + r.withErrorLogger(func(log Logger) { + log.Printf("kafka reader could not dial preferred read replica %d for partition %d of %s, falling back to leader: %v", r.preferredReadReplica, r.partition, r.topic, err) + }) + r.blockPreferredReplica(r.preferredReadReplica) + r.preferredReadReplica = -1 + r.preferredReadReplicaExpiresAt = time.Time{} + usePreferred = false + conn, err = r.dialer.DialLeader(ctx, "tcp", broker, r.topic, r.partition) + if err != nil { + continue + } + if first, last, err = r.readOffsets(conn); err != nil { + conn.Close() + conn = nil + break + } + } else { + // Concrete offset, no need to call readOffsets at all. + // The leader has already validated the offset for us on a + // prior fetch (we only got here because read() handed us a + // concrete offset to resume at). Use sentinel values that + // are inert in the clamping switch below. + first, last = 0, offset+1 + onFollower = true + r.connectedToFollower = true + } + } else { + t0 := time.Now() + if usePreferred && needOffsetResolution { + // Resolve markers against the leader first, then redial + // the preferred follower for the actual fetch. + leaderConn, lerr := r.dialer.DialLeader(ctx, "tcp", broker, r.topic, r.partition) + if lerr != nil { + r.stats.dials.observe(1) + r.stats.dialTime.observeDuration(time.Since(t0)) + err = lerr + continue + } + if first, last, err = r.readOffsets(leaderConn); err != nil { + leaderConn.Close() + r.stats.dials.observe(1) + r.stats.dialTime.observeDuration(time.Since(t0)) + break + } + leaderConn.Close() + + // Materialize the resolved offset so the rest of the loop + // treats this like a concrete-offset path. + switch offset { + case FirstOffset: + offset = first + case LastOffset: + offset = last + } - t0 := time.Now() - conn, err = r.dialer.DialLeader(ctx, "tcp", broker, r.topic, r.partition) - t1 := time.Now() - r.stats.dials.observe(1) - r.stats.dialTime.observeDuration(t1.Sub(t0)) - - if err != nil { - continue - } - - if first, last, err = r.readOffsets(conn); err != nil { - conn.Close() - conn = nil - break + // Now dial the preferred follower for the fetch path. + t1 := time.Now() + conn, err = r.dialPreferredReplica(ctx, broker) + r.stats.dials.observe(1) + r.stats.dialTime.observeDuration(time.Since(t1)) + if err != nil { + r.withErrorLogger(func(log Logger) { + log.Printf("kafka reader could not dial preferred read replica %d for partition %d of %s, falling back to leader: %v", r.preferredReadReplica, r.partition, r.topic, err) + }) + r.blockPreferredReplica(r.preferredReadReplica) + r.preferredReadReplica = -1 + r.preferredReadReplicaExpiresAt = time.Time{} + usePreferred = false + conn, err = r.dialer.DialLeader(ctx, "tcp", broker, r.topic, r.partition) + if err != nil { + continue + } + } else { + onFollower = true + r.connectedToFollower = true + } + } else { + conn, err = r.dialer.DialLeader(ctx, "tcp", broker, r.topic, r.partition) + r.stats.dials.observe(1) + r.stats.dialTime.observeDuration(time.Since(t0)) + if err != nil { + continue + } + if first, last, err = r.readOffsets(conn); err != nil { + conn.Close() + conn = nil + break + } + } } switch { @@ -1475,7 +1704,15 @@ func (r *reader) initialize(ctx context.Context, offset int64) (conn *Conn, star log.Printf("the kafka reader for partition %d of %s is seeking to offset %d", r.partition, r.topic, toHumanOffset(offset)) }) - if start, err = conn.Seek(offset, SeekAbsolute); err != nil { + // On a KIP-392 follower we must NOT issue ListOffsets to validate the + // seek — the follower will reject with NotLeaderForPartition. The + // offset has already been validated by the leader on the previous + // fetch, so use SeekDontCheck. + seekFlags := SeekAbsolute + if onFollower { + seekFlags |= SeekDontCheck + } + if start, err = conn.Seek(offset, seekFlags); err != nil { conn.Close() conn = nil break @@ -1501,6 +1738,64 @@ func (r *reader) read(ctx context.Context, offset int64, conn *Conn) (int64, err }) highWaterMark := batch.HighWaterMark() + // KIP-392: detect a (different) preferred read replica or expired + // preference. Update the cached state, but do NOT abandon the records + // the broker just gave us — that response is valid even if the broker + // wants us to talk to a different replica next time. We deliver the + // records first, then signal the run loop to reconnect. + prrChanged := false + if prr := batch.PreferredReadReplica(); prr != r.preferredReadReplica { + switch { + case prr < 0 && r.connectedToFollower && r.preferredReadReplica >= 0 && + time.Now().Before(r.preferredReadReplicaExpiresAt): + // We are currently fetching from the preferred follower and it + // reported "no further preference" (preferred_read_replica = -1 + // in the FetchResponse). Followers don't advertise themselves + // as their own preference, so this is the steady-state response + // and absolutely does NOT mean the leader has revoked our + // pinning. Treat it as a heartbeat: refresh the TTL and keep + // fetching from the same follower. Without this branch we + // would close the connection, reconnect to the leader, get + // re-redirected to the same follower, and oscillate every few + // hundred milliseconds until the broker's selector eventually + // stops re-asserting. + r.preferredReadReplicaExpiresAt = time.Now().Add(5 * time.Minute) + case prr >= 0 && r.preferredReplicaInCooldown(prr): + // If the broker is suggesting a replica we recently saw fail, + // ignore the suggestion and stay on the leader. Without this we + // would oscillate forever: the broker re-advertises the same id + // on every Fetch, and acting on it would just re-trigger the + // failure path. + // + // Treat the suggestion as "no preference" for our purposes. + // Do NOT touch r.preferredReadReplica here so that we don't + // trip the prrChanged signal on every batch. + default: + r.preferredReadReplica = prr + if prr >= 0 { + // Default to 5 minutes — matches Kafka's metadata.max.age.ms default + // and the reference behavior of the Java client for the preferred + // read replica TTL. + r.preferredReadReplicaExpiresAt = time.Now().Add(5 * time.Minute) + r.withLogger(func(log Logger) { + log.Printf("kafka reader switching to preferred read replica %d for partition %d of %s after current batch", prr, r.partition, r.topic) + }) + } else { + r.preferredReadReplicaExpiresAt = time.Time{} + r.withLogger(func(log Logger) { + log.Printf("kafka reader falling back to leader for partition %d of %s after current batch", r.partition, r.topic) + }) + } + prrChanged = true + } + } else if r.preferredReadReplica >= 0 && time.Now().After(r.preferredReadReplicaExpiresAt) { + // Preferred replica TTL elapsed — schedule a reconnect to the leader + // so we can re-discover whether the preference still applies. + r.preferredReadReplica = -1 + r.preferredReadReplicaExpiresAt = time.Time{} + prrChanged = true + } + t1 := time.Now() r.stats.waitTime.observeDuration(t1.Sub(t0)) @@ -1540,6 +1835,16 @@ func (r *reader) read(ctx context.Context, offset int64, conn *Conn) (int64, err r.stats.readTime.observeDuration(t2.Sub(t1)) r.stats.fetchSize.observe(size) r.stats.fetchBytes.observe(bytes) + + // If the broker wants us to switch replicas, reconnect AFTER having + // drained the batch. Only override the existing err if it is an + // expected end-of-batch signal (io.EOF or RequestTimedOut). Real + // errors (context cancellation, send failures, decode/connection + // errors) must propagate as-is so the run loop handles them. + if prrChanged && (err == nil || errors.Is(err, io.EOF) || errors.Is(err, RequestTimedOut)) { + conn.Close() + return offset, errPreferredReadReplicaChanged + } return offset, err } @@ -1619,3 +1924,58 @@ func (offset humanOffset) Format(w fmt.State, _ rune) { fmt.Fprint(w, strconv.FormatInt(v, 10)) } } + +// errPreferredReadReplicaChanged is an internal sentinel returned by reader.read +// to signal the outer run loop that we need to reconnect, either to a newly +// elected preferred read replica (KIP-392) or back to the leader after the +// preferred replica expired. +var errPreferredReadReplicaChanged = errors.New("preferred read replica changed") + +// dialPreferredReplica resolves r.preferredReadReplica's broker address by +// asking the seed broker for cluster metadata, then opens a partition +// connection to that broker. Returns an error if the broker id cannot be +// found or any dial step fails. +func (r *reader) dialPreferredReplica(ctx context.Context, seed string) (*Conn, error) { + // Open a short-lived connection to the seed broker just to look up + // brokers + the partition descriptor. + c, err := r.dialer.DialContext(ctx, "tcp", seed) + if err != nil { + return nil, fmt.Errorf("dial seed broker %s: %w", seed, err) + } + parts, err := c.ReadPartitions(r.topic) + c.Close() + if err != nil { + return nil, fmt.Errorf("read partitions: %w", err) + } + + var target Partition + var found bool + for _, p := range parts { + if p.ID != r.partition { + continue + } + // Only consider replicas that are actually in-sync. The leader is + // always in ISR and is a valid fallback target if the broker chose + // to point us at it. A broker present in Replicas but absent from + // Isr is offline or lagging and unsafe to follower-fetch from -- + // matches the Java client's behavior of skipping non-online + // replicas when applying KIP-392. + candidates := append([]Broker{p.Leader}, p.Isr...) + for _, b := range candidates { + if b.ID == int(r.preferredReadReplica) { + // Build a Partition view where the preferred replica is + // presented as the "leader" so DialPartition connects to it + // and the resulting Conn is bound to the right topic/partition. + target = p + target.Leader = b + found = true + break + } + } + break + } + if !found { + return nil, fmt.Errorf("preferred replica %d not in ISR for partition %d of %s", r.preferredReadReplica, r.partition, r.topic) + } + return r.dialer.DialPartition(ctx, "tcp", seed, target) +} diff --git a/write.go b/write.go index 3b806509c..63eab0ce9 100644 --- a/write.go +++ b/write.go @@ -295,6 +295,65 @@ func (wb *writeBuffer) writeFetchRequestV10(correlationID int32, clientID, topic return wb.Flush() } +// writeFetchRequestV11 is identical to writeFetchRequestV10 but additionally +// sends a rack id (KIP-392) which lets the broker direct the consumer to the +// closest replica. +func (wb *writeBuffer) writeFetchRequestV11(correlationID int32, clientID, topic string, partition int32, offset int64, minBytes, maxBytes int, maxWait time.Duration, isolationLevel int8, rackID string) error { + h := requestHeader{ + ApiKey: int16(fetch), + ApiVersion: int16(v11), + CorrelationID: correlationID, + ClientID: clientID, + } + h.Size = (h.size() - 4) + + 4 + // replica ID + 4 + // max wait time + 4 + // min bytes + 4 + // max bytes + 1 + // isolation level + 4 + // session ID + 4 + // session epoch + 4 + // topic array length + sizeofString(topic) + + 4 + // partition array length + 4 + // partition + 4 + // current leader epoch + 8 + // fetch offset + 8 + // log start offset + 4 + // partition max bytes + 4 + // forgotten topics data + sizeofString(rackID) // rack id + + h.writeTo(wb) + wb.writeInt32(-1) // replica ID + wb.writeInt32(milliseconds(maxWait)) + wb.writeInt32(int32(minBytes)) + wb.writeInt32(int32(maxBytes)) + wb.writeInt8(isolationLevel) + wb.writeInt32(0) // session ID + wb.writeInt32(-1) // session epoch + + // topic array + wb.writeArrayLen(1) + wb.writeString(topic) + + // partition array + wb.writeArrayLen(1) + wb.writeInt32(partition) + wb.writeInt32(-1) // current leader epoch + wb.writeInt64(offset) + wb.writeInt64(int64(0)) // log start offset only used when sent by follower + wb.writeInt32(int32(maxBytes)) + + // forgotten topics array + wb.writeArrayLen(0) + + // rack id (KIP-392) + wb.writeString(rackID) + + return wb.Flush() +} + func (wb *writeBuffer) writeListOffsetRequestV1(correlationID int32, clientID, topic string, partition int32, time int64) error { h := requestHeader{ ApiKey: int16(listOffsets),