...

Current state: Under Discussion

Depends On: KIP-1248

Discussion thread: [DISCUSS] KIP-1254: Kafka Consumer Support for Remote Tiered Storage Fetch

...

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

KIP-1248 introduces protocol changes to allow brokers to allow consumers to read remote log segments directly, reducing network bandwidth/cost and disk IOPs. 

...

The core record parsing logic exists in the clients module:

Consumer parsing flow:

1. Download byte range from remote storage via RemoteStorageFetcher
2. Use RemoteLogInputStream.nextBatch() to iterate over RecordBatch objects
3. Copy batches to ByteBuffer and wrap as MemoryRecords
4. Use existing CompletedFetch logic for transaction filtering

...

Share Groups (Queues)

Status: Not  Not supported in this KIP. Planned for a subsequent KIP.

Share groups require the broker to  to mediate record distribution via via SharePartition.acquire(), which manages complex state including record acquisition, timeouts, and acknowledgments. Direct consumer fetch is feasible in a future KIP where the broker uses OffsetIndex to determine byte ranges, returns offset metadata and segment location (instead of records), and consumers fetch record bytes directly from remote tiered storage. The transaction filtering logic in this KIP would extend to that design consumer fetch would require significant architectural changes to maintain this coordination.

This KIP focuses on standard consumer groups (AsyncKafkaConsumer/ClassicKafkaConsumer) for historical data reads. Share groups are typically used for real-time workloads rather than backfilling historical data, making this an edge case.

Fallback

The consumer falls back to broker-mediated fetch when remote fetch fails:

Version Compatibility

To handle storage format evolution, consumers include SupportedStorageFormatVersions in FetchRequest:

...

This also enables Kafka-API-compatible vendors with proprietary storage formats to participate (e.g. Apache Pulsar, WarpStreams, AutoMQ)

This design intentionally shifts segment parsing from the broker to the consumer to reduce broker load. While this couples the consumer to the on-disk format, SupportedStorageFormatVersions ensures graceful fallback when formats evolve. The consumer remains decoupled from storage backends via the RemoteStorageFetcher plugin interface.

Authorization

The security model is implementation-specific to each RemoteStorageFetcher plugin. We offer some thoughts on the different approaches, but this entirely depends on specific cloud-provider and is not specified in this KIP:

...

FetchRequest (version bump)

FetchResponse (version bump)

Note: existing abortedTransactions field will be populated by brokers for remote segments.

Consumer Configs

New Metrics

New consumer metrics will be added following the existing kafka.consumer:type=consumer-fetch-manager-metrics naming convention including:

...

Detailed metric definitions will be finalized during implementation.

Class Changes Summary

Migration Plan & Compatibility

Client-Broker Compatibility

Rollout Steps

1. Upgrade brokers to support KIP-1248
2. Deploy RemoteStorageFetcher implementation
3. Upgrade consumers with new client library
4. Enable fetch.remote.enabled=true per consumer group
5. Monitor remote-fetch-* metrics for issues

...

Disabling fetch.remote.enabled immediately returns to typical broker fetches. No data loss or protocol issues occur.

Rejected Alternatives