Kafka Fetch Sessions: Scaling Beyond Linear Requests

Your Kafka consumer, the client that reads records, owns thousands of partitions, ordered shards of a log, but only a few have data waiting. A sessionless Fetch request, the consumer's read command, still names every partition led by a broker, a Kafka server, on every round. The records are sparse; the control message isn't.

KIP-227 fixed that repetition by adding a fetch session: remembered partition state shared between a client and broker. The first request establishes the state. Later requests carry a delta, the partitions added, removed, or changed since the previous successful exchange.

TL;DR: KIP-227 changes incremental request metadata from O(P_b)—proportional to all partitions fetched from one broker—to O(Δ_b), proportional to changed partition entries. It doesn't make the entire fetch operation constant-time. The broker still checks the session's partitions, records still cross the network, and both endpoints now own session state. Fetch sessions pay off when Δ_b is much smaller than P_b, not merely when P_b is large.

By the end, you'll be able to:

• calculate what a sessionless Fetch request repeats;
• trace session creation, updates, closure, and recovery;
• account for the broker and client state that KIP-227 introduces;
• recognize workloads where incremental requests save little.

Without a session, every request restates the partition set

A consumer doesn't send one Fetch request to the whole cluster. Kafka groups assigned partitions by their current leader and sends a request to each destination broker. The Java client's AbstractFetch builds those broker-bound requests.

That distinction matters. The relevant count isn't every partition assigned to the consumer; it is P_b, the partitions fetched through one broker connection. The total repeated metadata across a fetch round is the sum of P_b over those connections.

Immediately before KIP-227, Fetch v6 grouped partitions by topic. Its request schema gave each partition four fixed-width fields:

A full request containing 10,000 partition rows therefore spends at least 240,000 bytes, about 234.4 KiB, on those rows alone. That isn't a benchmark. It is arithmetic from the v6 schema, before topic names, array lengths, and the request header.

The response had the same scaling shape. Before sessions, the broker returned a result for every requested partition, including partitions with no records. Those results still carried an error code and values such as the high watermark, the broker's replicated-read boundary. KIP-227 identifies both repeated request entries and empty response entries as costs that grow with partition count.

The narrow wording matters: KIP-227 reduces partition-description traffic and repeated serialization. It doesn't reduce the bytes occupied by records, and the broker still considers the effective partition set when looking for data. Put the O(Δ_b) label beside incremental metadata, not beside the entire Fetch operation.

The first full request buys the session

Fetch v7 added session_id, session_epoch, and forgotten_topics_data to the Fetch request schema. A client that wants a new session sends the complete partition set with session_id=0 and session_epoch=0.

The broker may place that set in its bounded session cache and return a non-zero session ID. If it can't admit a session, it returns ID 0; the exchange remains a full fetch. Either way, the initial request still contains P_b partition entries. The first round creates a baseline rather than saving bytes.

The following session ID is illustrative, but the transitions match the KIP-227 state machine:

① establishes the complete set. The broker has no earlier state it can use.

② sends only changed request state: p1 has a new fetch offset, p6 joined the set, and p5 left it. Partitions p0, p2, p3, and p4 remain in the broker's cached map without appearing in the request.

③ isn't restricted to partitions named in the request delta. The broker reconstructs the full set and may return records for an omitted partition if data arrived there.

Both session fields use fixed-width, signed INT32 values. They are not unsigned varints. FetchMetadata defines the complete lifecycle:

A delta describes changed fetch state, not future data

A common summary says incremental requests contain only partitions with new data. That reverses the timeline: the client sends its request before the broker reveals which partitions have records.

The Java FetchSessionHandler compares the desired fetch map with the map used for the session. It sends additions and partitions whose request parameters changed, while removals go into forgotten_topics_data.

A busy partition commonly appears in the next delta because consuming records advances its fetch offset. A quiet partition can disappear from requests for many rounds while remaining active in the broker's session.

Responses are incremental too. The broker's CachedPartition state remembers enough of the previous response to suppress unchanged, empty partition results. It still returns a partition when it has records, an error, or changed state such as a high watermark or log start offset. The Fetch response schema carries the session ID and any included partition responses.

An empty request delta doesn't mean the broker has no work. It means the client's desired fetch parameters haven't changed. The broker still checks the cached set and can return newly available records.

Epochs reject stale deltas; correlation IDs don't reorder them

An epoch is the sequence number for updates within one session. After a successful full exchange at epoch 0, the client sends incremental epochs in order. The broker rejects a value that doesn't match the session's expected epoch because applying a missing or duplicated delta would produce the wrong partition map.

Don't picture correlation IDs, request-header integers copied into responses, as keys for arbitrary response reordering. In Kafka 4.3.1, InFlightRequests.completeNext completes the oldest request queued for a connection. NetworkClient removes that request, and AbstractResponse validates the returned correlation ID.

Correlation IDs validate the expected pairing; the Java client doesn't search its queue for any request with the returned ID. Ordinary out-of-order responses therefore aren't the explanation for an invalid fetch-session epoch.

Recovery deliberately falls back to a full request

Incremental updates work only while both endpoints agree on the session ID, partition map, and next epoch. When that agreement breaks, Kafka doesn't try to merge two uncertain histories. It returns to a full request.

Kafka assigns dedicated protocol errors to the two main disagreements. The symbolic names and numeric values come from Kafka 4.3.1's Errors.java:

The Java client's recovery transitions live in FetchSessionHandler.handleResponse and handleError. A broker restart also loses the in-memory cache, while the disconnected client rebuilds from its complete local map.

A failed epoch doesn't necessarily erase the broker entry immediately. The client abandons that incremental chain; the bounded broker cache remains responsible for retiring stale entries. Recovery restores correctness, but its next request costs O(P_b) again.

What could go wrong: healthy sessions can churn after a broker roll

KAFKA-9137 documented a confirmed cache-maintenance bug after broker rolls. On a 12-broker cluster using the default 1,000 slots, eviction rates exceeded 16 sessions per second and an active replica-fetcher session could be evicted milliseconds after successful incremental Fetch requests. The next request then received FETCH_SESSION_ID_NOT_FOUND, forcing a full rebuild.

Kafka fixed that bug in 2.5.0. It remains a useful failure signature rather than a claim about 4.3.1: session evictions rise, IDs disappear, and clients repeatedly rebuild full state while record processing may continue. KAFKA-9401 later documented global cache-lock contention and led toward the sharded cache design used by current brokers.

Correctness fallback isn't the same as a healthy steady state. A consumer can keep moving records while request metadata, broker work, and session-error logs grow.

The optimization moves bytes into state at both endpoints

KIP-227 saves repeated wire bytes by retaining information that sessionless Fetch requests carried explicitly. That state has a cost on both sides.

Kafka 4.3 documents a default of 1000 for max.incremental.fetch.session.cache.slots. Current brokers divide that capacity across eight shards with shard-local eviction. A slot counts one session, not its partition entries or bytes. A session holding 20 partitions and one holding 20,000 partitions each consume one slot, but they don't retain the same amount of broker heap.

Consumer count isn't session count either. A consumer can maintain a separate session with each broker from which it fetches, and KIP-227 also covers replica fetchers. Cache demand follows active client-to-broker fetch paths and their partition maps.

The current broker exposes NumIncrementalFetchSessions, NumIncrementalFetchPartitionsCached, and IncrementalFetchSessionEvictionsPerSec from its fetch-session implementation. Read them together. A full slot count with stable partition totals tells a different story from continuously rising sessions, cached partitions, and evictions.

Opinion — protocol state is configuration debt. Fetch sessions save bytes while creating configuration debt: settings, metrics, lifecycle transitions, and debugging cases that didn't exist in the sessionless protocol. The trade is usually good for stable, high-partition fetchers, but calling the feature “automatic” hides the operational bill.

Raising the slot cap may reduce eviction while retaining more state. Lowering it bounds the cache while forcing more full rebuilds. There is no useful universal value because session count alone omits the number of partitions held by each session and their churn rate.

The Kafka 4.3.1 source makes the state machine reproducible

Protocol documentation explains the contract, but a pinned source release shows the exact client and broker transitions. Kafka's 4.3.1 tag is a useful inspection point because it shows the current sharded fetch-session cache without relying on the moving trunk branch.

The following commands reproduce the source trace. They inspect the wire fields, client state, broker cache, and response-ordering behavior discussed above.

git clone --branch 4.3.1 --depth 1 \
  https://github.com/apache/kafka.git kafka-4.3.1 # ①
cd kafka-4.3.1
 
git grep -n '"SessionId"\|"SessionEpoch"\|"ForgottenTopicsData"' -- \
  clients/src/main/resources/common/message/FetchRequest.json # ②
 
git grep -n \
  'class FetchSessionHandler\|class FetchSessionCacheShard\|class IncrementalFetchContext' # ③
 
git grep -n 'completeNext\|correlationId' -- \
  clients/src/main/java/org/apache/kafka/clients # ④

① pins every observation to the Kafka 4.3.1 release tag.

② shows that Fetch v7 introduced the session fields and that the schema types them as INT32.

③ leads to the client's delta builder and the broker's full, incremental, sessionless, and error contexts. Reading those classes side by side makes the duplicated state visible.

④ reaches the oldest-request completion and correlation-ID validation path. It is the reproducible check against the incorrect claim that Java clients accept arbitrary response order.

Fetch sessions stop helping when most partition state changes

High partition count is only half the fit. The better predictor is the ratio Δ_b / P_b: changed entries divided by all entries in the broker-bound session.

For an illustrative stable session with 10,000 partitions and 50 changed entries, the next request carries 200 times fewer partition rows. That doesn't imply a 200-times faster fetch: the broker still checks the session and sends whatever records are available.

At the opposite extreme, if nearly every partition returns records, nearly every fetch offset advances. The next delta then resembles the full map. Sessions remain correct, but the request-side byte saving largely disappears.

The useful question isn't “Do we have many partitions?” It is “How much of each broker-bound partition map changes between fetches?” Partition count creates the opportunity; stability determines whether KIP-227 captures it.

KIP-227 removed repetition, not linear work

The fetch-session state machine has five points worth keeping in a runbook:

• A sessionless Fetch request describes every assigned partition leader for one broker.
• The first session request is full; savings begin only after the broker accepts it.
• Incremental requests contain changed fetch parameters, not partitions predicted to have data.
• Session IDs and epochs are signed INT32 values, and the Java client validates responses against its oldest in-flight request per connection.
• Cache eviction, epoch disagreement, or connection ambiguity recovers through a full request.

KIP-227 didn't remove the linear partition list. It taught Kafka when not to resend it.

Next, the broker request path shows where that Fetch work waits after it reaches the server.

Sources and References

Protocol and KIP

• Incremental fetch design: KIP-227: Introduce Incremental FetchRequests to Increase Partition Scalability.
• Fetch request wire fields: FetchRequest.json at Kafka 4.3.1.
• Fetch response wire fields: FetchResponse.json at Kafka 4.3.1.
• Protocol reference: Apache Kafka 4.3 protocol documentation.

Kafka 4.3.1 Implementation

• Client session state and delta construction: FetchSessionHandler.java.
• Session constants and epoch transitions: FetchMetadata.java.
• Broker cache and fetch contexts: FetchSession.java.
• Per-broker consumer fetch construction: AbstractFetch.java.
• Oldest in-flight request completion: InFlightRequests.java.
• Correlation-ID validation: NetworkClient.java.
• Protocol error codes: Errors.java.
• Broker cache setting: max.incremental.fetch.session.cache.slots.

Historical failure reports

• Active-session eviction after broker rolls: KAFKA-9137.
• Global fetch-session-cache lock contention: KAFKA-9401.