Introduce client.writer.delivery-timeout to Fluss Write Client
- 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
Currently, the Fluss write client sets `client.writer.retries` to `Integer.MAX_VALUE` by default to support idempotent writes and fault-tolerant delivery. However, there is no overall delivery timeout governing how long a record or batch can remain in the client's accumulator and retry loop.
While `client.request.timeout` exists (default `30s`), it only limits individual RPC network roundtrips. When a TabletServer rejects writes due to a retriable condition (for example, `DiskWriteLockedException` when a data disk exceeds `server.data-disk.write-limit-ratio`, or transient cluster state changes like `LeaderNotAvailableException`), the server immediately returns an error response in `< 1 ms`. Because the RPC itself completes successfully, `client.request.timeout` is never triggered.
As a result:
1. `Sender.canRetry()` continuously re-enqueues the batch because `error.exception() instanceof RetriableException` is `true` and `attempts < Integer.MAX_VALUE`.
2. `TableWriter.flush()` invokes `RecordAccumulator.awaitFlushCompletion()`, which waits on an un-timed `CountDownLatch.await()`.
3. The client enters an **infinite silent retry loop**, causing user applications, integration tests, and calling threads to hang indefinitely without ever surfacing a `TimeoutException` or actionable error.
This gap has already been identified in the codebase with explicit placeholders (`RecordAccumulator.java`: `// TODO add deliveryTimeoutMs to report success or failure on record delivery.` and `// TODO add nextBatchExpiryTimeMs`, as well as `Sender.java`: `// TODO add logic for batch expire.`).
Following the precedent of [Apache Kafka KIP-91](https://cwiki.apache.org/confluence/display/KAFKA/KIP-91+-+Provide+Intuitive+User+Timeouts+in+The+Producer), we propose introducing `client.writer.delivery-timeout` along with accumulator expiration scheduling (`nextBatchExpiryTimeMs`) to place an upper bound on the entire batch lifecycle (queuing, in-flight RPCs, and all retries combined).
### Solution
_No response_
### Anything else?
_No response_
### Willingness to contribute
- [x] I'm willing to submit a PR!
Contributor guide
No contributing guide indexed for this repository
Research direction
Read RecordAccumulator.java and Sender.java, starting with the TODOs for deliveryTimeoutMs, nextBatchExpiryTimeMs, and batch expiration. Trace TableWriter.flush() through awaitFlushCompletion() and the retry path, then define completion as a bounded batch lifecycle that surfaces a TimeoutException instead of hanging indefinitely.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- backend-api-design, distributed-systems
- Issue type
- Feature
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Active
- Clarity
- Mostly clear
- Newbie friendliness
- 48/100