[Bug] Writer batches can hang forever after synchronous RPC encoding failure

[gyang94]

Search before asking

• I searched in the issues and found nothing similar.

Fluss version

main (development)

Please describe the bug 🐞

When the Fluss Sender encounters an OutOfMemoryError or another fatal synchronous error while encoding or sending a write RPC, some write batches can become permanently stuck.

The failure flow is:

1. The batches have already been drained from the accumulator and registered in Sender.inFlightBatches.
2. The RPC fails before it is actually sent to the server.
3. No response future or completion callback is available to handle these batches.
4. The batches are neither completed, re-enqueued, nor deallocated, so they continue holding writer-buffer memory indefinitely.
5. The writer buffer is eventually exhausted and subsequent writes block, while the Sender still appears to be running.

With idempotent writes, later batches may also receive OUT_OF_ORDER_SEQUENCE_EXCEPTION because an earlier sequence was never delivered or re-enqueued.

Root cause

1. Sender.sendWriteData drains ready batches from the accumulator.
2. It adds all drained batches to Sender.inFlightBatches before calling sendWriteRequests.
3. ServerConnection.doSend registers the RPC request before encoding it.
4. The encoding cleanup catches only Exception. A fatal Error, such as OutOfMemoryError, escapes synchronously from gateway.putKv(request) or gateway.produceLog(request).
5. Because the gateway call throws before returning a CompletableFuture, the Sender's whenComplete callback is never registered.
6. The outer Sender loop catches Throwable, logs it, and continues without completing, re-enqueuing, or deallocating the affected batches.

Expected behavior: every drained batch must either be associated with a completable RPC future, re-enqueued, completed successfully, or failed and deallocated. Fatal errors should fail the writer/task instead of only being logged.

Solution

• Clean up the RPC registration for every synchronous encoding failure.
• Ensure synchronous gateway failures cannot leave Sender batches orphaned.
• Stop the Sender and fail or abort incomplete batches on fatal errors.
• Add tests that inject a fatal encoding/send failure and verify that in-flight entries are removed, futures complete exceptionally, and writer-buffer memory is released.

Are you willing to submit a PR?

• I'm willing to submit a PR!