DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
Current state: Under Discussion
Depends On: KIP-1248
Discussion thread: [DISCUSS] KIP-1254: Kafka Consumer Support for Remote Tiered Storage Fetchhere [Change the link from the KIP proposal email archive to your own email thread]
JIRA: here [Change the link from KAFKA-1 to your own ticket]
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:
Class | Location | Purpose |
RemoteLogInputStream | Iterates RecordBatch from InputStream | |
MemoryRecords | Wraps ByteBuffer as usable records | |
CompletedFetch | Transaction filtering for read_committed | |
DefaultRecordBatch | Batch parsing (v2+ format) |
Consumer parsing flow:
- Download byte range from remote storage via RemoteStorageFetcher
- Use RemoteLogInputStream.nextBatch() to iterate over RecordBatch objects
- Copy batches to ByteBuffer and wrap as MemoryRecords
- 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:
Condition | Behavior |
Remote fetch timeout exceeded (fetch.remote.read.timeout.ms exceeded) | Re-issue FetchRequest with RemoteLogSegmentLocationRequested=false |
Connection timeout (fetch.remote.connect.timeout.ms exceeded) | Re-issue FetchRequest with RemoteLogSegmentLocationRequested=false |
Authentication failure | Re-issue FetchRequest with RemoteLogSegmentLocationRequested=false |
RemoteStorageFetcher not configured | Never set RemoteLogSegmentLocationRequested=true |
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)
Field | Type | Description |
RemoteLogSegmentLocationRequested | boolean | Whether consumer requests remote segment location |
SupportedStorageFormatVersions | []string | Storage format versions the consumer can parse |
FetchResponse (version bump)
Field | Type | Description |
RemoteLogSegmentId | UUID | Unique identifier of the remote segment |
RemoteLogSegmentCustomMetadata | bytes | Provider-specific metadata |
Note: existing abortedTransactions field will be populated by brokers for remote segments.
Consumer Configs
Config | Type | Default | Description |
fetch.remote.enabled | boolean | false | Controls whether consumers set RemoteLogSegmentLocationRequested=true |
remote.storage.fetcher.class | string | null | Implementation class for RemoteStorageFetcher interface |
remote.storage.fetcher.class.path | string | null | Classpath for loading RemoteStorageFetcher implementation |
int | 1000 | Connection timeout for remote storage. Triggers fallback on timeout | |
int | 5000 | Read timeout for remote storage. Triggers fallback on timeout |
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
Class | Change Type | Description |
RemoteStorageFetcher | New interface | Cloud-agnostic interface for fetching from remote storage |
RemoteStorageNetworkClient | New class | Routes requests to remote storage, converts responses |
SubscriptionState | Modified | Add RemoteLogSegmentId, RemoteLogSegmentCustomMetadata fields |
UnsentRequest | Modified | Add remote segment location fields |
FetchCollector | Modified | Extract remote segment location from FetchResponse |
Migration Plan & Compatibility
Client-Broker Compatibility
Scenario | Behavior |
New brokers, old consumers | Consumer never sets RemoteLogSegmentLocationRequested=true |
Old brokers, new consumers | Broker ignore the field, behaves as before |
Format mismatch | Broker checks SupportedStorageFormatVersions, falls back if incompatible |
Rollout Steps
- Upgrade brokers to support KIP-1248
- Deploy RemoteStorageFetcher implementation
- Upgrade consumers with new client library
- Enable fetch.remote.enabled=true per consumer group
- 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
Alternative | Reason for Rejection |
Broker-side filtering | Defeats the purpose of reducing broker load |
Reuse RemoteStorageManager | Too bulky for read-only client use, violates interface segregation |
Consumer downloads index files | Adds complexity; broker can provide needed info more efficiently |