swuferhong opened a new issue, #3262: URL: https://github.com/apache/fluss/issues/3262
### 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! -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
