feat: implement KIP-392 (consumer rack-aware fetching) - #1434
Open
GlebShipilov wants to merge 8 commits into
Open
feat: implement KIP-392 (consumer rack-aware fetching)#1434GlebShipilov wants to merge 8 commits into
GlebShipilov wants to merge 8 commits into
Conversation
Add RackID support to FetchRequest/FetchResponse and propagate PreferredReadReplica from the broker's Fetch v11+ response. This allows consumers to set a rack ID so the broker can direct them to fetch from the closest replica, reducing cross-datacenter traffic. Changes: - fetch.go: RackID field on FetchRequest, PreferredReadReplica on FetchResponse - reader.go: RackID config option on ReaderConfig, propagated through to fetch - conn.go: rack-aware fetch support in low-level Conn API - batch.go: expose PreferredReadReplica on Batch - dialer.go: RackID on DialerConfig - read.go: helper for rack-aware reading - write.go: pass-through support - protocol.go: PreferredReadReplica constant
feat: implement KIP-392 consumer rack awareness
The three-part form (1.23.0) is only understood by Go 1.21+ toolchains. The upstream segmentio/kafka-go CircleCI uses circleci/golang without a version tag, which pulls an older Go that rejects 1.23.0 with: invalid go version '1.23.0': must match format 1.23
preserve in-flight batch, leader fallback on follower errors, correct -1 default, ISR-only preferred-replica selection
…adOffsets failure When the Reader is connected to a preferred follower (KIP-392) and the follower returns NotLeaderForPartition (on fetch or during readOffsets in initialize), the stale preference was never cleared. The next initialize call dialed the same follower again, causing an infinite retry loop. Fix: - NotLeaderForPartition handler in readLoop: clear preferredReadReplica before breaking, same pattern as the existing OffsetOutOfRange fix. - initialize/readOffsets: when readOffsets fails on a preferred replica connection, clear the preference and retry via the leader on the same broker iteration instead of breaking out entirely.
…lock When the leader re-advertises the same PreferredReadReplica on every Fetch, a follower that rejects ListOffsets (NotLeaderForPartition) or fails to dial caused an infinite re-init loop: leader -> follower -> fail -> leader -> same hint -> follower -> ... This change adds a per-reader negative cache (preferredReadReplicaCooldown, 5m TTL matching the PRR TTL) keyed by broker id. Whenever a preferred replica fails (dial error, NotLeaderForPartition, OffsetOutOfRange) we record an "ignore until" timestamp. Subsequent Fetch responses that re-suggest the same broker id are ignored until the cooldown expires. Also: KIP-392 only allows Fetch on followers; ListOffsets stays on the leader. initialize() now resolves FirstOffset/LastOffset markers against the leader first and only then redials the preferred follower for the actual fetch. For a concrete resume offset we skip readOffsets entirely on the follower path.
The previous live-lock turned out to have a deeper root cause than the negative-cache alone could solve. After dialing a KIP-392 preferred follower and (correctly) skipping the explicit readOffsets call, initialize() still invoked conn.Seek(offset, SeekAbsolute). That call internally issues ReadOffsets (ListOffsets v?) against the connected broker to validate the requested offset — and a follower rejects it with NotLeaderForPartition. The error then surfaced as "error initializing the kafka reader for partition X: [6] Not Leader For Partition" on every retry, defeating the cooldown logic because the failure happened before any Fetch was ever issued. Fix: pass SeekAbsolute|SeekDontCheck whenever the connection we are about to seek on is a preferred follower. The leader has already validated the offset for us on the prior fetch that handed us this concrete resume offset, so the local seek is safe without a network round-trip to ListOffsets. A small `onFollower` flag is set in the two places where initialize() ends up holding a follower connection (the concrete-offset fast path and the redial-after-leader-resolution path) and consumed at the single conn.Seek call site.
After dialing a KIP-392 preferred follower, the very first FetchResponse
from that follower carries preferred_read_replica = -1, simply because
followers do not advertise themselves as their own preference. The
existing reader logic compared this against r.preferredReadReplica
(= the follower's broker id) and concluded the broker had revoked the
preference. It then closed the follower connection, re-dialed the
leader, got re-redirected to the same follower on the next fetch, and
oscillated for several seconds until the broker's selector eventually
stopped re-asserting the suggestion.
Symptom in the high-level consumer log:
routing_change preferred_replica_change -1 -> 103 (leader-driven)
routing_change ttl_or_no_preference 103 -> -1 (~800ms later)
... repeated for 5-10 partitions in the first ~10 seconds ...
Fix: split the prr-changed branch in reader.read() into three cases:
1. We are currently pinned to a follower AND the broker reported
prr=-1 AND the TTL is still valid -> treat as heartbeat, refresh
the 5-minute TTL, do NOT trip prrChanged. This is the common
steady-state path for follower fetches.
2. The broker is suggesting a replica that is in the local cooldown
(existing behavior) -> ignore.
3. Default -> existing behavior (switch / fall back / log).
A new connectedToFollower bool on the reader struct disambiguates
case 1 from "we are connected to the leader and it just told us to
fall back to itself". It is set in initialize() at the same two
points that already set the local onFollower flag for SeekDontCheck,
and reset on every initialize() call.
Net behavior:
- One preferred_replica_change per partition at startup (unchanged).
- Zero ttl_or_no_preference flips during steady state.
- Real TTL expiry every ~5 min still triggers a re-discovery cycle.
- All other prr transitions (cooldown, dial-fail, OffsetOutOfRange,
NotLeaderForPartition) take the existing paths unchanged.
Author
|
@petedannemann, can you have a look, please? |
|
Hi, we're running Would love to see this reviewed and merged. Happy to help test against MSK if useful. 🙏 |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What
Adds support for KIP-392 (follower fetching), which lets a consumer
fetch records from the closest in-rack replica instead of always going
to the partition leader. This cuts cross-AZ traffic and tail latency in
multi-rack Kafka clusters.
Changes
KIP-392 core (wire format + routing)
fetch.go: addedRackIDfield onFetchRequest,PreferredReadReplicafield onFetchResponse.reader.go: addedRackoption onReaderConfig. When set, thereader passes it through to every fetch.
dialer.go: addedRackIDonDialerso the low-levelConnAPIcan be rack-aware too.
conn.go: handles the v11 fetch response header, decodes thePreferredReadReplicafield, and exposes it onBatch.batch.go: newPreferredReadReplica()accessor onBatch.read.go,write.go,protocol.go: small plumbing changes to carrythe new field through the wire format.
protocol/fetch/fetch.go: taggedRackIDon the request andPreferredReadReplicaon the response per partition for v11+.Drain in-flight batch before reconnecting
When the
PreferredReadReplicain a fetch response differs from thecached value, upstream would immediately close the batch and connection,
discarding unread records. This PR sets a
prrChangedflag, deliversevery record from the batch first, then signals
errPreferredReadReplicaChangedto the run loop. The reconnect happens between batches, not mid-batch.
ISR-only replica selection
dialPreferredReplica()resolves the preferred replica againstp.Isr(not
p.Replicas). A broker that is in the replica set but out of ISR —because it is offline or lagging — is skipped. This matches the Java
client's behavior and prevents fetching from a stale follower.
Clear preferred replica on
NotLeaderForPartitionWhen a
NotLeaderForPartitionerror arrives while connected to apreferred follower, the reader now clears the preference and adds the
replica to the cooldown cache before falling back to the leader.
Without this,
initialize()would re-dial the same failed followerin a tight loop.
Clear preferred replica on follower
OffsetOutOfRangeAn
OffsetOutOfRangefrom a follower is not authoritative becausefollower offsets can lag behind the leader's. The reader now drops
the preference and lets
initialize()reconnect through the leader,whose first/last offsets are the only ones trusted for the reset
decision.
Negative cache for failed preferred replicas
A per-reader
preferredReadReplicaCooldownmap records an "ignoreuntil" timestamp for each recently-failed replica. Subsequent fetch
responses that re-advertise the same broker id are silently ignored
until the 5-minute cooldown expires. Without this the broker
re-suggests the same dead replica on every fetch and the reader
live-locks in a dial→fail→fallback→re-suggest loop.
SeekDontCheckon follower connectionsIn
initialize(), when the reader is about to seek on a KIP-392follower, it uses
SeekAbsolute|SeekDontCheckinstead of plainSeekAbsolute. This skips theListOffsetsvalidation call, whichfollowers reject with
NotLeaderForPartition(only leaders serveListOffsets). The offset has already been validated by the leaderon the prior fetch.
Resolve
FirstOffset/LastOffsetvia leader before follower dialWhen the reader needs to resolve
FirstOffsetorLastOffset(e.g.after a group rebalance), it now dials the leader first to call
ReadOffsets, materializes the concrete offset, then re-dials thepreferred follower for the actual fetch path. This avoids sending
ListOffsetsto the follower.Follower's
PreferredReadReplica = -1is a heartbeatPer KIP-392,
-1from a follower the reader is already pinned to means"stay with me" (the follower does not advertise itself as its own
preference).
-1from the leader means "no preferred replica, stay onme." The reader now tracks
connectedToFollowerto disambiguate thesetwo cases: when connected to the follower,
-1refreshes the TTLinstead of triggering a reconnect. Without this the reader oscillates
between leader and follower every few hundred milliseconds.
How it was tested
Ran the full test matrix locally against Kafka 2.7.0, 2.8.1, and 3.7.0
(the same versions CircleCI uses):
go test -race -cover ./...go test -tags=unsafe -race -cover ./...cd sasl/aws_msk_iam && go test -race -cover ./...All tests pass except
TestReaderConsumerGroup, which is a pre-existingnil-pointer panic on
mainunrelated to this change.Chaos testing on a live cluster
Ran a structured chaos protocol against a 10-broker, 3-rack Strimzi
cluster (KRaft mode, RF=3, MIRR=2) with a
go-kafka-consumerhighlevelconsumer reading 10 partitions. The protocol covers every failure mode
the resilience fixes are designed for:
NotLeaderForPartitionclearing, KIP-392 proof pointResults: zero
Not Leader For Partitionerrors, zerofetch_message_failed, zero routing oscillations, all partitionsresumed committing within 60 s of each kill. Full report in
KAFKA_OUTAGE_TEST_REPORT.md.Notes on the
PreferredReadReplica = -1edge caseThis is the part that's easy to get wrong. Per KIP-392, a value of
-1means different things depending on which broker sent it:
-1from the leader means "no preferred replica for you, stay on theleader" — the client must drop any cached follower.
-1from a follower that the client is already cached on means"stay with me" — the client must keep the cache.
If both cases are treated the same, the consumer oscillates between
leader and follower on every fetch and the optimization is silently
broken. The
connectedToFollowerflag disambiguates the two cases,matching the Java client behavior.
Backwards compatibility
The new fields default to zero values, so consumers that don't set
RackIDsee no behavior change. Brokers older than Fetch v11 simplyignore the field. All resilience mechanisms are inert when
preferredReadReplicais-1(the default).