You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@fluss.apache.org by Sagar <sa...@gmail.com> on 2026/08/24 02:33:19 UTC
[DISCUSS] FIP-54: Introduce client.writer.delivery-timeout to Fluss Write Client
Hi,
I created FIP-54:
https://cwiki.apache.org/confluence/spaces/FLUSS/pages/449285308/FIP-54+Introduce+client.writer.delivery-timeout+to+Fluss+Write+Client
to add delivery timeouts in Fluss writes.
Please take a look.
Sagar.
Re: [DISCUSS] FIP-54: Introduce client.writer.delivery-timeout to Fluss Write Client
Posted by Sagar <sa...@gmail.com>.
Hello,
Please let me know if my responses are making sense. I can update the fip
accordingly and if there are no further questions, can open the fip for
voting.
Sagar.
On Mon, 28 Sep 2026 at 8:39 PM, Zhe Wang <ja...@gmail.com> wrote:
> Hi Sagar,
>
> Thanks for the detailed response. Independent expiry of in-flight batches
> addresses the RPC-budget question, and the explicit flush() contract and
> direct Java API test address my second question.
>
> I have one remaining question about how the expiry scan keeps running.
> Currently, Sender.runOnce() calls maybeWaitForWriterId(), which waits on
> initWriter(...).get() without a timeout. This path is also entered after
> resetWriterId(). If that RPC future remains incomplete, Sender cannot
> perform further expiry scans, so queued batches could still remain pending
> beyond their delivery deadline.
>
> Could the FIP clarify how expiry will continue during writer-id allocation
> or reacquisition, and during synchronous metadata refreshes? When updating
> the sleep calculation, the earliest deadline should also include in-flight
> batches, including when the accumulator is empty.
>
> A test where the InitWriter future never completes, but record futures
> still fail at the delivery deadline, would help verify this part of the
> guarantee. It would be useful to cover both initial allocation and
> reacquisition after a writer-id reset.
>
> Best regards,
> Zhe Wang
>
>
> Sagar <sa...@gmail.com> 于2026年9月28日周一 01:44写道：
>
> > Hi Giannis/Zhe,
> >
> > Thank you for the in depth review. Please find my response below:
> >
> > * ### Idempotence & Sequence Numbers*
> >
> > In Fluss, batchSequence is assigned lazily when a batch is drained from
> > RecordAccumulator (batch.setWriterState(writerId, nextSequence)). This
> > leads to two scenarios:
> >
> > - *Queued batches in the accumulator*: Batches that expire before
> being
> > drained have never been assigned a sequence number
> > (!batch.hasBatchSequence()). Aborting them creates no sequence gap,
> and
> > subsequent batches continue with contiguous sequence numbers without
> > issue.
> > - *In-flight or retrying batches*: These batches already hold an
> > assigned sequence number. If such a batch expires, dropping it would
> > leave
> > a sequence gap and cause subsequent batches to fail with
> > OutOfOrderSequenceException. Following Kafka KIP-91 and Fluss's
> existing
> > design in IdempotenceManager.resetWriterId() (which already notes: "or
> > the
> > batch expired in writer queue after it was already sent to the tablet
> > server"), expiring an already-sequenced batch will trigger
> > resetWriterId().
> > The client will obtain a new writerId (or bump epoch) so subsequent
> > batches
> > restart from sequence 0, safely preserving order without fatal
> sequence
> > errors.
> >
> > * ### Batch Expiry and in-flight batches*
> > * 1. Expiring in-flight batches:*
> >
> > - Yes, in-flight batches will be expired when the delivery deadline is
> > reached. To Giannis's question: the timer starts as soon as a batch is
> > created on the first append, and batches will indeed expire and be
> > dropped
> > if they exceed delivery-timeout while waiting in the accumulator.
> > - As in KIP-91, delivery.timeout.ms represents an end-to-end bound
> from
> > batch creation. The background Sender will scan both the accumulator
> > queue
> > and inFlightBatches. If now - batch.createdMs >= deliveryTimeoutMs,
> the
> > batch will be failed immediately with a TimeoutException, notifying
> the
> > user without waiting for the in-flight RPC to complete.
> > - This ensures delivery.timeout.ms does provide a strict upper bound
> on
> > queuing, in-flight RPCs, and retries combined, even in cases where
> Netty
> > request deadlines are delayed (see
> > https://github.com/apache/fluss/issues/4279).
> >
> >
> > * 2. Handling late responses:*
> >
> > - Fluss’s WriteBatch already contains the atomic state machine needed
> > for this (WriteBatch.done() uses compareAndSet(null, tryFinalState)).
> > - If a batch is aborted due to delivery timeout, any late response
> from
> > the tablet server is discarded and logged at DEBUG.
> > - Because the server may or may not have committed this batch,
> > idempotenceManager.resetWriterId() will be invoked to acquire a new
> > writer
> > ID / epoch, ensuring surviving batches restart from sequence 0 without
> > encountering OutOfOrderSequenceException (which we saw in the
> > Idempotence
> > point as well).
> >
> > * 3. Bounding RPC timeouts by remaining budget:*
> >
> > - We will enforce the configuration invariant at startup:
> > client.writer.delivery-timeout >= client.writer.batch-timeout +
> > client.request-timeout + client.writer.retry-backoff. This ensures
> every
> > batch has enough budget for at least one full attempt.
> > - We do not dynamically clamp the per-RPC request timeout for retries.
> > Because Produce requests batch multiple buckets together, clamping
> would
> > couple disparate batches. Instead, expiring in-flight batches
> > independently
> > (Point 1) already provides the strict delivery deadline guarantee to
> the
> > caller.
> >
> > *## *TimeoutException is retriable-typed* (extends RetriableException). *
> >
> > That's a good point and I missed this. To get around this, we can add a
> > short-circuiting check at the top of Sender.canRetry() which basically
> > checks if the delivery time has elapsed for this batch. An added benefit
> of
> > doing this is that it also protects against other transient exceptions
> like
> > DisconnectException or LeaderNotAvailableException. Again, IIUC, this
> > mechanism is similar to the KIP-91 where a similar function called
> > `hasReachedDeliveryTimeout` performs the short-circuiting.
> >
> > ## Backward Compatibility & Metrics
> >
> > • Compatibility: Agreed — while binary-compatible, this is a behavioral
> > change (outages that previously caused indefinite blocking will now fail
> > after the timeout). I will remove "fully backward compatible" and add a
> > prominent upgrade note.
> > • Metrics: I will add an expiredRecordsTotal counter metric to
> > WriterMetricGroup and track batch queue/delivery latency.
> >
> > *## flush() public Contract*
> >
> > Yes, that is the intended public contract. In Fluss, flush() is designed
> > strictly as a synchronization barrier ensuring all previously buffered
> > records have completed—meaning they have either been acknowledged or
> > reached a terminal failure. Propagating exceptions directly from flush()
> > would be ambiguous when flushing multiple batches with mixed outcomes, so
> > failures (including TimeoutException) are delivered through the
> individual
> > record futures/callbacks where callers can determine the outcome of each
> > record. I agree with your suggestion and will update FIP-54 to explicitly
> > state this contract.
> > As suggested, I will also add a direct test for Java TableWriter API, not
> > only the Flink sink path.
> >
> >
> > If these responses seem ok, I will update the FIP and hopefully it will
> be
> > in a position where we can start voting on it. Please let me know!
> >
> > Sagar.
> >
> >
> > On Wed, Sep 23, 2026 at 10:37 PM Zhe Wang <ja...@gmail.com> wrote:
> >
> > > Hi Sagar,
> > >
> > > Thanks for proposing this. The motivation makes sense to me, especially
> > > since the current unlimited retry behavior can block flush
> indefinitely.
> > >
> > > I have two questions about the proposed semantics and implementation:
> > >
> > > 1. Does delivery-timeout actually bound in-flight RPCs?
> > >
> > > The proposed abortExpiredBatches() scans batches still in the
> > accumulator,
> > > while the hasExpired() check in canRetry() is only reached after an RPC
> > > returns an error.
> > >
> > > However, client.request-timeout currently does not enforce an
> independent
> > > client-side RPC deadline. As described in
> > > https://github.com/apache/fluss/issues/4279, a request that receives
> > > neither a response nor a connection failure may remain incomplete until
> > the
> > > connection is closed.
> > >
> > > This means an in-flight batch could still remain pending beyond
> > > client.writer.delivery-timeout. Even after client-side request
> deadlines
> > > are introduced, a retry started near the delivery deadline could exceed
> > the
> > > overall budget unless its request deadline is capped by the remaining
> > > delivery time, or the batch is expired independently while in flight.
> > >
> > > Could the FIP clarify:
> > >
> > > - whether in-flight batches are expired when the delivery deadline is
> > > reached;
> > > - how late responses are handled after a batch has already expired;
> > > - and whether each RPC timeout should be bounded by the batch's
> remaining
> > > delivery budget?
> > >
> > > Otherwise, I think the proposal should avoid claiming that it places a
> > > strict upper bound on queuing, in-flight RPCs, and retries combined.
> > >
> > > 2. How should flush() expose an expired batch?
> > >
> > > Currently, RecordAccumulator.awaitFlushCompletion() only waits for each
> > > RequestFuture's CountDownLatch. RequestFuture does not retain or expose
> > the
> > > batch exception, so expiring a batch will unblock WriterClient.flush(),
> > but
> > > flush() itself may still return normally. The TimeoutException is only
> > > delivered through the individual record futures or callbacks.
> > >
> > > Is that the intended public contract?
> > >
> > > If so, I think the FIP should state clearly that flush() only waits for
> > > completion and callers must inspect the record futures for failures.
> > > Otherwise, flush() may need to propagate the first or an aggregated
> > > exception from the records covered by that flush.
> > >
> > > It would also be useful to add a test for the direct Java TableWriter
> > API,
> > > not only the Flink sink path, since the Flink sink explicitly checks
> > > asynchronous write failures after flush while ordinary client users may
> > > rely on flush() alone.
> > >
> > > Best regards,
> > > Zhe Wang
> > >
> > >
> > > Sagar <sa...@gmail.com> 于2026年8月24日周一 10:34写道：
> > >
> > > > Hi,
> > > >
> > > > I created FIP-54:
> > > >
> > > >
> > >
> >
> https://cwiki.apache.org/confluence/spaces/FLUSS/pages/449285308/FIP-54+Introduce+client.writer.delivery-timeout+to+Fluss+Write+Client
> > > >
> > > > to add delivery timeouts in Fluss writes.
> > > >
> > > > Please take a look.
> > > >
> > > > Sagar.
> > > >
> > >
> >
>
Re: [DISCUSS] FIP-54: Introduce client.writer.delivery-timeout to Fluss Write Client
Posted by Zhe Wang <ja...@gmail.com>.
Hi Sagar,
Thanks for the detailed response. Independent expiry of in-flight batches
addresses the RPC-budget question, and the explicit flush() contract and
direct Java API test address my second question.
I have one remaining question about how the expiry scan keeps running.
Currently, Sender.runOnce() calls maybeWaitForWriterId(), which waits on
initWriter(...).get() without a timeout. This path is also entered after
resetWriterId(). If that RPC future remains incomplete, Sender cannot
perform further expiry scans, so queued batches could still remain pending
beyond their delivery deadline.
Could the FIP clarify how expiry will continue during writer-id allocation
or reacquisition, and during synchronous metadata refreshes? When updating
the sleep calculation, the earliest deadline should also include in-flight
batches, including when the accumulator is empty.
A test where the InitWriter future never completes, but record futures
still fail at the delivery deadline, would help verify this part of the
guarantee. It would be useful to cover both initial allocation and
reacquisition after a writer-id reset.
Best regards,
Zhe Wang
Sagar <sa...@gmail.com> 于2026年9月28日周一 01:44写道：
> Hi Giannis/Zhe,
>
> Thank you for the in depth review. Please find my response below:
>
> * ### Idempotence & Sequence Numbers*
>
> In Fluss, batchSequence is assigned lazily when a batch is drained from
> RecordAccumulator (batch.setWriterState(writerId, nextSequence)). This
> leads to two scenarios:
>
> - *Queued batches in the accumulator*: Batches that expire before being
> drained have never been assigned a sequence number
> (!batch.hasBatchSequence()). Aborting them creates no sequence gap, and
> subsequent batches continue with contiguous sequence numbers without
> issue.
> - *In-flight or retrying batches*: These batches already hold an
> assigned sequence number. If such a batch expires, dropping it would
> leave
> a sequence gap and cause subsequent batches to fail with
> OutOfOrderSequenceException. Following Kafka KIP-91 and Fluss's existing
> design in IdempotenceManager.resetWriterId() (which already notes: "or
> the
> batch expired in writer queue after it was already sent to the tablet
> server"), expiring an already-sequenced batch will trigger
> resetWriterId().
> The client will obtain a new writerId (or bump epoch) so subsequent
> batches
> restart from sequence 0, safely preserving order without fatal sequence
> errors.
>
> * ### Batch Expiry and in-flight batches*
> * 1. Expiring in-flight batches:*
>
> - Yes, in-flight batches will be expired when the delivery deadline is
> reached. To Giannis's question: the timer starts as soon as a batch is
> created on the first append, and batches will indeed expire and be
> dropped
> if they exceed delivery-timeout while waiting in the accumulator.
> - As in KIP-91, delivery.timeout.ms represents an end-to-end bound from
> batch creation. The background Sender will scan both the accumulator
> queue
> and inFlightBatches. If now - batch.createdMs >= deliveryTimeoutMs, the
> batch will be failed immediately with a TimeoutException, notifying the
> user without waiting for the in-flight RPC to complete.
> - This ensures delivery.timeout.ms does provide a strict upper bound on
> queuing, in-flight RPCs, and retries combined, even in cases where Netty
> request deadlines are delayed (see
> https://github.com/apache/fluss/issues/4279).
>
>
> * 2. Handling late responses:*
>
> - Fluss’s WriteBatch already contains the atomic state machine needed
> for this (WriteBatch.done() uses compareAndSet(null, tryFinalState)).
> - If a batch is aborted due to delivery timeout, any late response from
> the tablet server is discarded and logged at DEBUG.
> - Because the server may or may not have committed this batch,
> idempotenceManager.resetWriterId() will be invoked to acquire a new
> writer
> ID / epoch, ensuring surviving batches restart from sequence 0 without
> encountering OutOfOrderSequenceException (which we saw in the
> Idempotence
> point as well).
>
> * 3. Bounding RPC timeouts by remaining budget:*
>
> - We will enforce the configuration invariant at startup:
> client.writer.delivery-timeout >= client.writer.batch-timeout +
> client.request-timeout + client.writer.retry-backoff. This ensures every
> batch has enough budget for at least one full attempt.
> - We do not dynamically clamp the per-RPC request timeout for retries.
> Because Produce requests batch multiple buckets together, clamping would
> couple disparate batches. Instead, expiring in-flight batches
> independently
> (Point 1) already provides the strict delivery deadline guarantee to the
> caller.
>
> *## *TimeoutException is retriable-typed* (extends RetriableException). *
>
> That's a good point and I missed this. To get around this, we can add a
> short-circuiting check at the top of Sender.canRetry() which basically
> checks if the delivery time has elapsed for this batch. An added benefit of
> doing this is that it also protects against other transient exceptions like
> DisconnectException or LeaderNotAvailableException. Again, IIUC, this
> mechanism is similar to the KIP-91 where a similar function called
> `hasReachedDeliveryTimeout` performs the short-circuiting.
>
> ## Backward Compatibility & Metrics
>
> • Compatibility: Agreed — while binary-compatible, this is a behavioral
> change (outages that previously caused indefinite blocking will now fail
> after the timeout). I will remove "fully backward compatible" and add a
> prominent upgrade note.
> • Metrics: I will add an expiredRecordsTotal counter metric to
> WriterMetricGroup and track batch queue/delivery latency.
>
> *## flush() public Contract*
>
> Yes, that is the intended public contract. In Fluss, flush() is designed
> strictly as a synchronization barrier ensuring all previously buffered
> records have completed—meaning they have either been acknowledged or
> reached a terminal failure. Propagating exceptions directly from flush()
> would be ambiguous when flushing multiple batches with mixed outcomes, so
> failures (including TimeoutException) are delivered through the individual
> record futures/callbacks where callers can determine the outcome of each
> record. I agree with your suggestion and will update FIP-54 to explicitly
> state this contract.
> As suggested, I will also add a direct test for Java TableWriter API, not
> only the Flink sink path.
>
>
> If these responses seem ok, I will update the FIP and hopefully it will be
> in a position where we can start voting on it. Please let me know!
>
> Sagar.
>
>
> On Wed, Sep 23, 2026 at 10:37 PM Zhe Wang <ja...@gmail.com> wrote:
>
> > Hi Sagar,
> >
> > Thanks for proposing this. The motivation makes sense to me, especially
> > since the current unlimited retry behavior can block flush indefinitely.
> >
> > I have two questions about the proposed semantics and implementation:
> >
> > 1. Does delivery-timeout actually bound in-flight RPCs?
> >
> > The proposed abortExpiredBatches() scans batches still in the
> accumulator,
> > while the hasExpired() check in canRetry() is only reached after an RPC
> > returns an error.
> >
> > However, client.request-timeout currently does not enforce an independent
> > client-side RPC deadline. As described in
> > https://github.com/apache/fluss/issues/4279, a request that receives
> > neither a response nor a connection failure may remain incomplete until
> the
> > connection is closed.
> >
> > This means an in-flight batch could still remain pending beyond
> > client.writer.delivery-timeout. Even after client-side request deadlines
> > are introduced, a retry started near the delivery deadline could exceed
> the
> > overall budget unless its request deadline is capped by the remaining
> > delivery time, or the batch is expired independently while in flight.
> >
> > Could the FIP clarify:
> >
> > - whether in-flight batches are expired when the delivery deadline is
> > reached;
> > - how late responses are handled after a batch has already expired;
> > - and whether each RPC timeout should be bounded by the batch's remaining
> > delivery budget?
> >
> > Otherwise, I think the proposal should avoid claiming that it places a
> > strict upper bound on queuing, in-flight RPCs, and retries combined.
> >
> > 2. How should flush() expose an expired batch?
> >
> > Currently, RecordAccumulator.awaitFlushCompletion() only waits for each
> > RequestFuture's CountDownLatch. RequestFuture does not retain or expose
> the
> > batch exception, so expiring a batch will unblock WriterClient.flush(),
> but
> > flush() itself may still return normally. The TimeoutException is only
> > delivered through the individual record futures or callbacks.
> >
> > Is that the intended public contract?
> >
> > If so, I think the FIP should state clearly that flush() only waits for
> > completion and callers must inspect the record futures for failures.
> > Otherwise, flush() may need to propagate the first or an aggregated
> > exception from the records covered by that flush.
> >
> > It would also be useful to add a test for the direct Java TableWriter
> API,
> > not only the Flink sink path, since the Flink sink explicitly checks
> > asynchronous write failures after flush while ordinary client users may
> > rely on flush() alone.
> >
> > Best regards,
> > Zhe Wang
> >
> >
> > Sagar <sa...@gmail.com> 于2026年8月24日周一 10:34写道：
> >
> > > Hi,
> > >
> > > I created FIP-54:
> > >
> > >
> >
> https://cwiki.apache.org/confluence/spaces/FLUSS/pages/449285308/FIP-54+Introduce+client.writer.delivery-timeout+to+Fluss+Write+Client
> > >
> > > to add delivery timeouts in Fluss writes.
> > >
> > > Please take a look.
> > >
> > > Sagar.
> > >
> >
>
Re: [DISCUSS] FIP-54: Introduce client.writer.delivery-timeout to Fluss Write Client
Posted by Sagar <sa...@gmail.com>.
Hi Giannis/Zhe,
Thank you for the in depth review. Please find my response below:
* ### Idempotence & Sequence Numbers*
In Fluss, batchSequence is assigned lazily when a batch is drained from
RecordAccumulator (batch.setWriterState(writerId, nextSequence)). This
leads to two scenarios:
- *Queued batches in the accumulator*: Batches that expire before being
drained have never been assigned a sequence number
(!batch.hasBatchSequence()). Aborting them creates no sequence gap, and
subsequent batches continue with contiguous sequence numbers without issue.
- *In-flight or retrying batches*: These batches already hold an
assigned sequence number. If such a batch expires, dropping it would leave
a sequence gap and cause subsequent batches to fail with
OutOfOrderSequenceException. Following Kafka KIP-91 and Fluss's existing
design in IdempotenceManager.resetWriterId() (which already notes: "or the
batch expired in writer queue after it was already sent to the tablet
server"), expiring an already-sequenced batch will trigger resetWriterId().
The client will obtain a new writerId (or bump epoch) so subsequent batches
restart from sequence 0, safely preserving order without fatal sequence
errors.
* ### Batch Expiry and in-flight batches*
* 1. Expiring in-flight batches:*
- Yes, in-flight batches will be expired when the delivery deadline is
reached. To Giannis's question: the timer starts as soon as a batch is
created on the first append, and batches will indeed expire and be dropped
if they exceed delivery-timeout while waiting in the accumulator.
- As in KIP-91, delivery.timeout.ms represents an end-to-end bound from
batch creation. The background Sender will scan both the accumulator queue
and inFlightBatches. If now - batch.createdMs >= deliveryTimeoutMs, the
batch will be failed immediately with a TimeoutException, notifying the
user without waiting for the in-flight RPC to complete.
- This ensures delivery.timeout.ms does provide a strict upper bound on
queuing, in-flight RPCs, and retries combined, even in cases where Netty
request deadlines are delayed (see
https://github.com/apache/fluss/issues/4279).
* 2. Handling late responses:*
- Fluss’s WriteBatch already contains the atomic state machine needed
for this (WriteBatch.done() uses compareAndSet(null, tryFinalState)).
- If a batch is aborted due to delivery timeout, any late response from
the tablet server is discarded and logged at DEBUG.
- Because the server may or may not have committed this batch,
idempotenceManager.resetWriterId() will be invoked to acquire a new writer
ID / epoch, ensuring surviving batches restart from sequence 0 without
encountering OutOfOrderSequenceException (which we saw in the Idempotence
point as well).
* 3. Bounding RPC timeouts by remaining budget:*
- We will enforce the configuration invariant at startup:
client.writer.delivery-timeout >= client.writer.batch-timeout +
client.request-timeout + client.writer.retry-backoff. This ensures every
batch has enough budget for at least one full attempt.
- We do not dynamically clamp the per-RPC request timeout for retries.
Because Produce requests batch multiple buckets together, clamping would
couple disparate batches. Instead, expiring in-flight batches independently
(Point 1) already provides the strict delivery deadline guarantee to the
caller.
*## *TimeoutException is retriable-typed* (extends RetriableException). *
That's a good point and I missed this. To get around this, we can add a
short-circuiting check at the top of Sender.canRetry() which basically
checks if the delivery time has elapsed for this batch. An added benefit of
doing this is that it also protects against other transient exceptions like
DisconnectException or LeaderNotAvailableException. Again, IIUC, this
mechanism is similar to the KIP-91 where a similar function called
`hasReachedDeliveryTimeout` performs the short-circuiting.
## Backward Compatibility & Metrics
• Compatibility: Agreed — while binary-compatible, this is a behavioral
change (outages that previously caused indefinite blocking will now fail
after the timeout). I will remove "fully backward compatible" and add a
prominent upgrade note.
• Metrics: I will add an expiredRecordsTotal counter metric to
WriterMetricGroup and track batch queue/delivery latency.
*## flush() public Contract*
Yes, that is the intended public contract. In Fluss, flush() is designed
strictly as a synchronization barrier ensuring all previously buffered
records have completed—meaning they have either been acknowledged or
reached a terminal failure. Propagating exceptions directly from flush()
would be ambiguous when flushing multiple batches with mixed outcomes, so
failures (including TimeoutException) are delivered through the individual
record futures/callbacks where callers can determine the outcome of each
record. I agree with your suggestion and will update FIP-54 to explicitly
state this contract.
As suggested, I will also add a direct test for Java TableWriter API, not
only the Flink sink path.
If these responses seem ok, I will update the FIP and hopefully it will be
in a position where we can start voting on it. Please let me know!
Sagar.
On Wed, Sep 23, 2026 at 10:37 PM Zhe Wang <ja...@gmail.com> wrote:
> Hi Sagar,
>
> Thanks for proposing this. The motivation makes sense to me, especially
> since the current unlimited retry behavior can block flush indefinitely.
>
> I have two questions about the proposed semantics and implementation:
>
> 1. Does delivery-timeout actually bound in-flight RPCs?
>
> The proposed abortExpiredBatches() scans batches still in the accumulator,
> while the hasExpired() check in canRetry() is only reached after an RPC
> returns an error.
>
> However, client.request-timeout currently does not enforce an independent
> client-side RPC deadline. As described in
> https://github.com/apache/fluss/issues/4279, a request that receives
> neither a response nor a connection failure may remain incomplete until the
> connection is closed.
>
> This means an in-flight batch could still remain pending beyond
> client.writer.delivery-timeout. Even after client-side request deadlines
> are introduced, a retry started near the delivery deadline could exceed the
> overall budget unless its request deadline is capped by the remaining
> delivery time, or the batch is expired independently while in flight.
>
> Could the FIP clarify:
>
> - whether in-flight batches are expired when the delivery deadline is
> reached;
> - how late responses are handled after a batch has already expired;
> - and whether each RPC timeout should be bounded by the batch's remaining
> delivery budget?
>
> Otherwise, I think the proposal should avoid claiming that it places a
> strict upper bound on queuing, in-flight RPCs, and retries combined.
>
> 2. How should flush() expose an expired batch?
>
> Currently, RecordAccumulator.awaitFlushCompletion() only waits for each
> RequestFuture's CountDownLatch. RequestFuture does not retain or expose the
> batch exception, so expiring a batch will unblock WriterClient.flush(), but
> flush() itself may still return normally. The TimeoutException is only
> delivered through the individual record futures or callbacks.
>
> Is that the intended public contract?
>
> If so, I think the FIP should state clearly that flush() only waits for
> completion and callers must inspect the record futures for failures.
> Otherwise, flush() may need to propagate the first or an aggregated
> exception from the records covered by that flush.
>
> It would also be useful to add a test for the direct Java TableWriter API,
> not only the Flink sink path, since the Flink sink explicitly checks
> asynchronous write failures after flush while ordinary client users may
> rely on flush() alone.
>
> Best regards,
> Zhe Wang
>
>
> Sagar <sa...@gmail.com> 于2026年8月24日周一 10:34写道：
>
> > Hi,
> >
> > I created FIP-54:
> >
> >
> https://cwiki.apache.org/confluence/spaces/FLUSS/pages/449285308/FIP-54+Introduce+client.writer.delivery-timeout+to+Fluss+Write+Client
> >
> > to add delivery timeouts in Fluss writes.
> >
> > Please take a look.
> >
> > Sagar.
> >
>
Re: [DISCUSS] FIP-54: Introduce client.writer.delivery-timeout to Fluss Write Client
Posted by Zhe Wang <ja...@gmail.com>.
Hi Sagar,
Thanks for proposing this. The motivation makes sense to me, especially
since the current unlimited retry behavior can block flush indefinitely.
I have two questions about the proposed semantics and implementation:
1. Does delivery-timeout actually bound in-flight RPCs?
The proposed abortExpiredBatches() scans batches still in the accumulator,
while the hasExpired() check in canRetry() is only reached after an RPC
returns an error.
However, client.request-timeout currently does not enforce an independent
client-side RPC deadline. As described in
https://github.com/apache/fluss/issues/4279, a request that receives
neither a response nor a connection failure may remain incomplete until the
connection is closed.
This means an in-flight batch could still remain pending beyond
client.writer.delivery-timeout. Even after client-side request deadlines
are introduced, a retry started near the delivery deadline could exceed the
overall budget unless its request deadline is capped by the remaining
delivery time, or the batch is expired independently while in flight.
Could the FIP clarify:
- whether in-flight batches are expired when the delivery deadline is
reached;
- how late responses are handled after a batch has already expired;
- and whether each RPC timeout should be bounded by the batch's remaining
delivery budget?
Otherwise, I think the proposal should avoid claiming that it places a
strict upper bound on queuing, in-flight RPCs, and retries combined.
2. How should flush() expose an expired batch?
Currently, RecordAccumulator.awaitFlushCompletion() only waits for each
RequestFuture's CountDownLatch. RequestFuture does not retain or expose the
batch exception, so expiring a batch will unblock WriterClient.flush(), but
flush() itself may still return normally. The TimeoutException is only
delivered through the individual record futures or callbacks.
Is that the intended public contract?
If so, I think the FIP should state clearly that flush() only waits for
completion and callers must inspect the record futures for failures.
Otherwise, flush() may need to propagate the first or an aggregated
exception from the records covered by that flush.
It would also be useful to add a test for the direct Java TableWriter API,
not only the Flink sink path, since the Flink sink explicitly checks
asynchronous write failures after flush while ordinary client users may
rely on flush() alone.
Best regards,
Zhe Wang
Sagar <sa...@gmail.com> 于2026年8月24日周一 10:34写道：
> Hi,
>
> I created FIP-54:
>
> https://cwiki.apache.org/confluence/spaces/FLUSS/pages/449285308/FIP-54+Introduce+client.writer.delivery-timeout+to+Fluss+Write+Client
>
> to add delivery timeouts in Fluss writes.
>
> Please take a look.
>
> Sagar.
>
Re: [DISCUSS] FIP-54: Introduce client.writer.delivery-timeout to Fluss Write Client
Posted by Giannis Polyzos <ip...@gmail.com>.
I probably responded to the wrong thread, so I’m adding my response to the
proper one 😄
First of all thanks for initiating this.
The motivation makes sense, and I verified it against the code: retries
defaults to Integer.MAX_VALUE, Sender.canRetry() re-enqueues indefinitely
on any RetriableException, and flush() bottoms out in an untimed
CountDownLatch.await().
Three things I'd want I wanted to point out here:
1. *Idempotence + sequence numbers*. Idempotence defaults on. Expiring a
batch that already holds a sequence leaves a gap, and the next batch fails
fatally with *OutOfOrderSequenceException*, malforming the bucket. KIP-91
co-designed delivery-timeout with producer-id/epoch handling for this
reason. I think we should specify what happens to writer-id/sequence state
on expiry and how ordering is preserved for surviving batches.
2. *In-flight vs. queued expiry*. The motivating case
(DiskWriteLockedException, acks=all) is a slow in-flight batch the server
may persist; failing it to the user risks an app-level retry and a
duplicate. We should state explicitly whether we expire in-flight batches
or only queued ones, and the trade-off either way.
3. *TimeoutException is retriable-typed* (extends RetriableException). The
expiry path must be terminal and bypass canRetry()/reEnqueue(), and
canRetry() needs a hasExpired() short-circuit; otherwise, the batch we're
expiring loops forever.
NIT: I'd not call this "fully backward-compatible". Behaviorally, a
transient outage that's survived by blocking today becomes a write failure
(and a Flink job restart) under a 2-minute default. We can do something
similar to Kafka, but we should also include a prominent upgrade note.
Also, we could add an expired-records counter alongside
recordsRetryTotal/recordsSendTotal, since the point is surfacing
previously-silent failures.
Let me know your thoughts.
Best,
Giannis
On Sun, 30 Aug 2026 at 7:02 PM, Sagar <sa...@gmail.com> wrote:
> Hi,
>
> Bumping this thread (though I think the team is busy with 1,0 release so I
> can wait not a problem).
>
> Sagar.
>
> On Mon, Aug 24, 2026 at 8:03 AM Sagar <sa...@gmail.com> wrote:
>
> > Hi,
> >
> > I created FIP-54:
> >
> >
> https://cwiki.apache.org/confluence/spaces/FLUSS/pages/449285308/FIP-54+Introduce+client.writer.delivery-timeout+to+Fluss+Write+Client
> >
> > to add delivery timeouts in Fluss writes.
> >
> > Please take a look.
> >
> > Sagar.
> >
>
Re: [DISCUSS] FIP-54: Introduce client.writer.delivery-timeout to Fluss Write Client
Posted by Sagar <sa...@gmail.com>.
Hi,
Bumping this thread (though I think the team is busy with 1,0 release so I
can wait not a problem).
Sagar.
On Mon, Aug 24, 2026 at 8:03 AM Sagar <sa...@gmail.com> wrote:
> Hi,
>
> I created FIP-54:
>
> https://cwiki.apache.org/confluence/spaces/FLUSS/pages/449285308/FIP-54+Introduce+client.writer.delivery-timeout+to+Fluss+Write+Client
>
> to add delivery timeouts in Fluss writes.
>
> Please take a look.
>
> Sagar.
>