Versions Compared

Old Version 1

changes.mady.by.user Thomas Thornton

Saved on

compared with

New Version 2

changes.mady.by.user Thomas Thornton

Saved on

Key

  • This line was added.
  • This line was removed.
  • Formatting was changed.

Authors: Thomas Thornton, Henry Cai

Depends on: KIP-1248

Table of Contents

Status

Current state: Under Discussion

...

The core record parsing logic exists in the clients module:

Class

Location

Purpose

RemoteLogInputStream

clients/…/common/record/

Iterates RecordBatch from InputStream

MemoryRecords

clients/…/common/record/

Wraps ByteBuffer as usable records

CompletedFetch

clients/…/consumer/internals/

Transaction filtering for read_committed

DefaultRecordBatch

clients/…/common/record/

Batch parsing (v2+ format)

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

...

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:

...

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 

fetch.remote.connect.timeout.ms

int

1000

Connection timeout for remote storage. Triggers fallback on timeout

fetch.remote.read.timeout.ms

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

  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

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