apache / apache/fluss

[client] Refactor RemoteLogDownloader to use chunked streaming instead of downloading whole segment file

Open
#3,262 0 comments 0 reactions 1 assignee Claimed by @swuferhong View on GitHub
Dominant language
Java
Stars
2.1k
Forks
625
Avg merge
3d 14h
Merged PRs (30d)
97

Description

### Search before asking

- [x] I searched in the [issues](https://github.com/apache/fluss/issues) and found nothing similar.

### Motivation

The current `RemoteLogDownloader` downloads the entire remote log segment file to local disk before any data can be consumed. This has several problems:

- Wasted download bandwidth: If the consumer only needs a portion of a segment (e.g., it unsubscribes, seeks, or closes mid-segment), the remaining bytes have already been downloaded and are simply discarded. For large segments (hundreds of MBs), this wastes significant remote storage bandwidth — which is especially costly for cloud object storage (OSS, S3) where egress is metered.

- High time-to-first-byte latency: The consumer must wait for the full segment to finish downloading before it can start processing any records.

- No flow control at chunk level: The only flow control is the prefetchSemaphore which limits the number of active segments, but does not limit how far ahead data is downloaded within a single segment. This means data continues to be eagerly downloaded even when the consumer is slow, further amplifying unnecessary bandwidth usage.

### Solution

Refactor `RemoteLogDownloader` to read remote log segments in configurable chunks (default 8 MB) instead of downloading the whole file at once:

- Introduce `RemoteSegmentChunkReader` to read fixed-size chunks from the remote FSDataInputStream, aligning on batch boundaries.

- Each chunk is appended to a local temp file and returned as a `FileLogRecords` slice (leveraging OS page cache, no JVM heap copy).

- Add two-level flow control:

1. prefetchSemaphore: limits the number of concurrently active segments (existing, default 4).

1. maxPrefetchChunks: limits the number of unconsumed chunks per segment (new, default 5). The downloader pauses when this limit is reached and resumes only when chunks are consumed.

- Use a continuationQueue to prioritize chunk continuation over new segment downloads.

- Chain chunk futures via nextChunkCallback so that LogFetcher automatically registers `RemotePendingFetch` for each subsequent chunk.

- New config options: `client.scanner.remote-log.chunk-size` (default 8 MB), `client.scanner.remote-log.max-prefetch-chunks` (default 5).

### Anything else?

_No response_

### Willingness to contribute

- [x] I'm willing to submit a PR!

Contributor guide

No contributing guide indexed for this repository

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.