[c++] Add write callbacks with batch completion dispatch

[naivedogger]

Purpose

Linked issue: close #4276

Add C++ write callbacks so applications can handle write completion without maintaining a separate WriteResult waiting queue.

Brief change log

• Add callback overloads for Append, AppendArrowBatch, Upsert, and Delete.
• Register callbacks directly on Rust write batches and dispatch them through shared workers, avoiding per-record async waiting tasks.
• Add coverage for registration races, error propagation, result ordering, and callback lifetime and exceptions.

Tests

• Rust core: 794 tests passed; Rust C++ bindings: 9 tests passed.
• Existing C++ test binary: 4 non-cluster tests passed.
• cargo fmt, cargo clippy, and git diff --check passed.
• No live-cluster tests or full Maven verification for this revision.

API and Format

• Add C++ callback overloads while preserving existing Wait/await result semantics and retry behavior. Shared completion allocation and notification scheduling change.
• Return submission status synchronously and report completion through callbacks. Submission errors do not register callbacks; multi-bucket ArrowBatch writes remain non-atomic.
• Callbacks may execute concurrently and out of order. The completion queue is unbounded, and Flush/destruction do not drain callbacks.
• The existing Rust shutdown-retry issue, delivery deadlines, and durable recovery are outside this PR's scope.
• No wire or storage format changes.

Documentation

Update the C++ API reference with callback examples, compatibility notes, partial-failure semantics, and application responsibilities for memory limits and shutdown.

[naivedogger]

@fresh-borzoni @loserwang1024 @leekeiabstraction, Appreciate a review here, thanks! 🙏

[fresh-borzoni]

@naivedogger Thank you, left a comment, PTAL

[naivedogger]

@fresh-borzoni thanks for the earlier review. I pushed three commits on top. enqueue_timeout now bounds the whole callback submission, both callback capacity and buffer backpressure, in the max.block.ms style, and a zero timeout makes submission non-blocking. The callback executor thread count is now overridable through an advanced FLUSS_CALLBACK_WORKERS variable, with the default of four unchanged.

I also expanded the docs and example around callback behavior: callbacks run on a small shared pool and should stay short and non-blocking, failures should be recorded and handled outside the callback with dedup by id, and crash recovery should come from a replayable source that advances only after Flush. When you have time, another look would be appreciated.

[loserwang1024]

pushed three commits on top. enqueue_timeout now bounds the whole callback submission, both callback capacity and buffer backpressure,

I do think client.writer.buffer.wait-timeout is enough, no need top. enqueue_timeout anymore

[naivedogger]

@loserwang1024 I think you’re right. I’ve updated it to use client.writer.buffer.wait-timeout.

[fresh-borzoni]

@naivedogger Thank you, LGTM 👍

[loserwang1024]

I have left some advice

[loserwang1024]

On WriteCallbackOptions.max_pending_operations

I'd like to confirm my understanding of the intended role of max_pending_operations — overall the design makes sense to me, but a few questions:

1. Relationship with client.writer.buffer.memory-size

From reading the code: memory permits are held until the batch completes (ACK), while max_pending_operations is reserved at submission and released only after the callback finishes executing on a worker. So the window it covers is really "after ACK, while the completion is queued or executing on the shared executor" — where captures are held in the (intentionally unbounded) completion queue with no other accounting.

So is the main failure mode it guards against slow user callbacks (submissions outpacing the 4-worker consumption rate, so callbacks pile up), rather than data volume? It would help to state this explicitly in the docs: memory-size bounds buffered bytes, max_pending_operations bounds outstanding callback operations (captures), and neither substitutes for the other.

2. Why does only C++ need this?

Neither the Java nor the Rust client has an equivalent. My reading is that this is inherent to this design rather than an oversight:

• Java returns a CompletableFuture and the SDK's responsibility ends at future.complete() — callback scheduling runs on a user-supplied executor, so the SDK never queues callbacks internally;
• Here the dispatch happens on the IO path, so the binding hosts a shared executor with an unbounded queue (a slow callback must not block IO), and an unbounded internal queue then requires admission control at the source.

Is that the intent? As a follow-up (not necessarily for this PR): would you consider an overload that lets the caller inject their own executor (e.g., Notify(cb, executor) or an executor field in WriteCallbackOptions), so advanced users can own the queuing the way Java users do and bypass the shared pool?

3. Sizing guidance for production

Do you have recommended values? Two related observations:

• The default 262144 is per writer, while the executor and its queue are process-wide, so with N writers the worst-case backlog is N × max.
• More importantly, the count-based limit and the byte-based limit don't derive from each other: with a small average record size (or a large writer_buffer_memory_size), the number of in-flight records the buffer can hold far exceeds 262144 — e.g. 512 MiB / 200 B ≈ 2.6M — so a callback submission can hit the capacity limit long before the buffer is full, and Append blocks in Acquire (bounded by wait-timeout) even though plenty of buffer memory is still free.

Is that "capacity binds before memory" behavior intended, and is it documented? And could the docs include a concrete derivation, e.g.:

• lower bound: max_pending_operations ≳ peak throughput × (ack latency + callback execution time / workers) so the limit never binds under normal load;
• upper bound: max_pending_operations × average capture size ≤ callback memory budget.

The "Sizing callback capacity and write buffers" section hints at this, but an explicit formula would make it much easier to configure.

[naivedogger]

Removed the public capacity option and now derive the internal callback limit from the writer buffer and schema estimate.

[loserwang1024]

why need this in this PR? If you want to introduce deadline, maybe another PR (and do it in both java and rust client)

[naivedogger]

Kept this only as doc-hidden binding plumbing to share the existing writer buffer wait budget, without adding a public deadline API.

[loserwang1024]

This is a blocking concern for me: callbacks for the same bucket must fire in completion order. I can't accept out-of-order notification within a single bucket, regardless of whether it's documented as intended behavior.

Worth spelling out where the ordering is actually lost, because it sharpens the point:

1. Same-bucket batches already complete in order, end to end: per-bucket in-flight is serialized, so batch2 isn't even sent until batch1 is ACKed.
2. ACKs on the same connection are processed in order, so the completion job for batch1 is enqueued before batch2's.
3. The channel itself is FIFO.

So ordering is destroyed in exactly one place: multiple workers dequeuing in parallel. The server, the sender, and the queue all preserve order — the executor is the first and only component to throw it away. Users then pay for it with locks plus order-defensive logic (advance by max, never by assignment) in exactly the most common callback patterns: per-bucket state and offset/checkpoint tracking. For reference, Kafka users get serialized onCompletion on the sender thread, and Fluss Java users get non-async thenAccept on the completing thread by default — this PR hands C++ users strictly less than both. (A late Notify on an already-completed batch enqueues at registration time — that's user-side causality and orthogonal to completion ordering.)

Ask for this PR: default to a single dispatch worker.

• FIFO queue + single consumer = callbacks fire in completion order (globally, which subsumes per-bucket). The change is small: worker count → 1. Chunking can stay or be simplified — it only existed to spread work across workers.
• The throughput concern has a clean answer, and it should become part of the contract: callbacks are notifications, not work items. The docs should state explicitly that callbacks must be lightweight — no heavy compute or IO; if you need that, hand off to your own thread pool. This is exactly the guidance Kafka gives for onCompletion, and the situation here is strictly milder than Kafka's: a slow callback only delays other callbacks — it never touches the IO path or write throughput (the queue is unbounded by design, and max_pending_operations bounds the backlog).

Evolution path (no API change): if dispatch parallelism is ever needed, shard the executor — key by writer or bucket id, one queue and one worker per key. That restores per-key parallelism while keeping per-key FIFO, and Notify/WriteCallbackOptions don't change. Until there's a demonstrated need, I'd keep it simple.

Related: with a single worker, queue drain time becomes the dominant term in AwaitAll's budget, so the hardcoded 60s in Flush() — and its error conflating "flushed successfully" with "callbacks didn't drain" — becomes easier to trigger. Please fix that alongside whichever direction we take here.

[naivedogger]

Changed dispatch to one FIFO worker, added a same-bucket ordering test, and made Flush wait on an unbounded callback barrier.

[naivedogger]

@loserwang1024 Thanks for your thorough review! I have pushed an update to address your points, could you please take another look?

[fresh-borzoni]

@naivedogger looked through as well, left some comments, PTAL

Not from your PR, but followup issue in the same path:

AppendArrowBatch sends one bucket group at a time. If a send fails partway, the earlier groups are already queued and get written, but the caller only sees an erro and resending duplicates those rows. Idempotence does not help her as it dedupes internal retries, not a fresh append.

The plain path does the same, and Java has no multi-bucket append so it is unaffected.

I would propose two steps in a separate issue. Make Err mean nothing was accepted, returning Ok with a joined future carrying the error when some groups went through. Then surface per-bucket results: PbProduceLogRespForBucket already carries bucket_id, error_code and base_offset, but BatchWriteResult is Result<(), Error> and drops them, so a caller cannot retry only the buckets that failed.

Separately, +1 to @loserwang1024 's injectable executor idea. C++'s default is already better than Java's here since it's ordered and off the I/O thread, but Java users can opt out with thenAcceptAsync and C++ users have no equivalent. It would also hand the queue bound to whoever owns the executor, which removes the need to guess max_pending_operations.

made a thorough pass over the code one more time

[naivedogger]

@fresh-borzoni Thanks for the review. I've pushed an update with WriteCompletion, containing only Result for now so we can add write-specific metadata later. You're right about the schema-based capacity estimate. I've restored a fixed default with a configurable per-writer limit. This limits outstanding callback operations, independently of write-buffer bytes. Capacity is reserved before submitting a write, using the existing buffer wait-timeout. Buffer release doesn't wait for user callbacks, although slow callbacks can fill the callback capacity and make later callback submissions wait or time out.

I've kept the single callback worker, fixed the late-registration ordering race, and removed the parallel delivery fallback. I've also corrected the stale four-worker documentation, updated the examples, and removed the timed AwaitAll overload and adjusted its tests. Flush() waits for callbacks without a separate hardcoded timeout.

For ArrowBatch, I agree with handling this separately in the two steps you suggested. First, return a synchronous Err only when nothing was accepted; otherwise retain the accepted groups' completion handles and report the partial-submission failure through the returned future. Then add per-bucket results so callers can identify individual outcomes instead of retrying the whole batch blindly. I'll open an issue to track that work, and another for user-provided callback executors, keeping the current single-worker default in this PR.

[loserwang1024]

LGTM, @fresh-borzoni @leonardBang , would you like to give a final check.

[fresh-borzoni]

+1, I'll wait until CI is green to merge