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

fluss-rust/bindings/cpp/BUILD.bazel

@@ -55,6 +55,9 @@ genrule(
     name = "cargo_build_debug",
     srcs = glob([
         "src/**/*.rs",
+        "src/**/*.hpp",
+        "include/**/*.hpp",
+        "build.rs",
         "Cargo.toml",
     ]),
     outs = [
@@ -121,6 +124,9 @@ genrule(
     name = "cargo_build_release",
     srcs = glob([
         "src/**/*.rs",
+        "src/**/*.hpp",
+        "include/**/*.hpp",
+        "build.rs",
         "Cargo.toml",
     ]),
     outs = [
@@ -274,6 +280,7 @@ cc_library(
     textual_hdrs = [
         "src/ffi_converter.hpp",
         "src/type_lowering.hpp",
+        "src/write_callback.hpp",
         ":rust_bridge_h_unified",
         ":lib_rs_h_unified",
         ":cxx_h_unified",

fluss-rust/bindings/cpp/README.md

@@ -65,7 +65,7 @@ not apply to `CreateBucketBatchScanner()`.
 
 ## Examples and Documentation
 
-- [examples/example.cpp](examples/example.cpp) demonstrates log-table writes, continuous scans,
+- [examples/example.cpp](examples/example.cpp) demonstrates log-table writes with Wait and bounded callbacks, continuous scans,
   bounded Arrow record-batch scans, projections, and offset queries.
 - [examples/admin_example.cpp](examples/admin_example.cpp) demonstrates database, table,
   partition, and cluster administration.
@@ -76,6 +76,33 @@ not apply to `CreateBucketBatchScanner()`.
   [C++ API reference](../../website/docs/user-guide/cpp/api-reference.md) and
   [log-table examples](../../website/docs/user-guide/cpp/example/log-tables.md).
 
+The callback section configures `WriteCallbackOptions` to bound outstanding callback
+operations. The SDK executes callbacks; applications do not need a waiting thread or
+poll loop. The example uses the default `max_pending_operations = 262144` per Writer.
+A callback submit is bounded by `client.writer.buffer.wait-timeout`, covering callback
+capacity plus buffer backpressure, so it returns a definite result within that budget
+and a zero timeout makes the submit non-blocking. It does not bound ACKs, retries, or
+callback duration.
+
+The callback limit counts operations, not bytes; it does not preallocate 262144 slots.
+The separate `Configuration::writer_buffer_memory_size` remains 64 MiB by default,
+shared across all tables and writers on a Connection. Neither setting caps process RSS.
+For a high-throughput starting configuration, see the
+[buffer sizing guidance](../../website/docs/user-guide/cpp/api-reference.md#sizing-callback-capacity-and-write-buffers),
+including a 512 MiB per-Connection example and how to budget for multiple writers.
+
+A failed callback does not prove that the record was not written. Application
+resubmission can duplicate it, even with SDK idempotence enabled. The example only
+counts and logs outcomes; it does not implement durable recovery. Keep callbacks
+short, protect shared state, and handle retries outside the callback with an
+application recovery policy.
+
+After submissions stop, `Flush()` first flushes writes and, on success, waits up to
+60 seconds for pending callbacks. This is not a whole-call timeout. If it returns an
+error, keep callback state alive; if it succeeds, still check individual write results.
+See the [callback guarantees and recovery guidance](../../website/docs/user-guide/cpp/api-reference.md#write-guarantees-and-recovery)
+for result semantics, callback implementation, and shutdown requirements.
+
 For a bounded log scan, pass the per-bucket offset ranges directly to `TableScan`. The returned
 reader yields one Arrow batch at a time until every `[starting_offset, stopping_offset)` range
 is complete:

fluss-rust/bindings/cpp/build.rs

@@ -17,8 +17,12 @@
 
 fn main() {
     cxx_build::bridge("src/lib.rs")
+        .include("include")
+        .include("src")
         .std("c++17")
         .compile("fluss-cpp-bridge");
 
     println!("cargo:rerun-if-changed=src/lib.rs");
+    println!("cargo:rerun-if-changed=src/write_callback.hpp");
+    println!("cargo:rerun-if-changed=include/fluss.hpp");
 }

fluss-rust/bindings/cpp/examples/example.cpp

@@ -20,9 +20,15 @@
 #include <arrow/record_batch.h>
 #include <arrow/type.h>
 
+#include <atomic>
 #include <chrono>
+#include <condition_variable>
+#include <exception>
 #include <iostream>
+#include <memory>
+#include <mutex>
 #include <unordered_map>
+#include <utility>
 #include <vector>
 
 #include "fluss.hpp"
@@ -39,6 +45,11 @@ int main() {
     // 1) Connect
     fluss::Configuration config;
     config.bootstrap_servers = "127.0.0.1:9123";
+    // Bounds how long a write blocks when the shared write buffer is full, and also caps
+    // the whole callback submission (callback capacity plus buffer backpressure). The
+    // default UINT64_MAX waits indefinitely; a finite value makes overloaded writes and
+    // callback submits return with an error instead of blocking. Zero fails fast.
+    config.writer_buffer_wait_timeout_ms = 30000;
 
     fluss::Connection conn;
     check("create", fluss::Connection::Create(config, conn));
@@ -85,7 +96,15 @@ int main() {
 
     // 5) Write rows with scalar and temporal values
     fluss::AppendWriter writer;
-    check("new_append_writer", table.NewAppend().CreateWriter(writer));
+    // Callback admission limit, shown with its default. It bounds the callback overloads
+    // only; Wait and fire-and-forget writes are unaffected. Passing no options uses this
+    // same value, so this block is equivalent to CreateWriter(writer).
+    fluss::WriteCallbackOptions callback_options;
+    // Pending callback operations per writer. Lower it for large captures or many
+    // writers; independent of the Connection's write-buffer byte budget. Waiting for a
+    // free slot is bounded by the connection's writer_buffer_wait_timeout_ms set above.
+    callback_options.max_pending_operations = 262144;
+    check("new_append_writer", table.NewAppend().CreateWriter(writer, callback_options));
 
     struct RowData {
         int id;
@@ -143,6 +162,60 @@ int main() {
         std::cout << "Row acknowledged by server" << std::endl;
     }
 
+    // Callback acknowledgment
+    {
+        // The SDK runs callbacks; no application waiting thread is required.
+        // Callbacks run on a small shared executor pool, so keep them short and
+        // non-blocking: do not Flush/Wait or retry synchronously inside a callback.
+        // This example counts outcomes only. It does not implement durable recovery.
+        std::atomic<size_t> succeeded{0};
+        std::atomic<size_t> failed{0};
+        for (const auto& r : rows) {
+            const int32_t id = 1000 + r.id;
+            fluss::GenericRow row;
+            row.SetInt32(0, id);
+            row.SetString(1, r.name);
+            row.SetFloat32(2, r.score);
+            row.SetInt32(3, r.age);
+            row.SetDate(4, r.date);
+            row.SetTime(5, r.time);
+            row.SetTimestampNtz(6, r.ts_ntz);
+            row.SetTimestampLtz(7, r.ts_ltz);
+            auto submitted = writer.Append(row, [id, &succeeded, &failed](fluss::Result result) {
+                if (result.Ok()) {
+                    ++succeeded;
+                } else {
+                    // By now the SDK has exhausted internal retries or hit a
+                    // non-retriable error, so do not retry synchronously here.
+                    // An error does not prove the row was not written, and a new
+                    // Append can duplicate it even with SDK idempotence enabled.
+                    // Record the outcome and either stop or hand id and input to
+                    // your own retry queue; deduplicate by id downstream.
+                    if (failed.fetch_add(1) == 0) {
+                        std::cerr << "Write failed for id=" << id << ": " << result.error_message
+                                  << '\n';
+                    }
+                }
+            });
+            if (!submitted.Ok()) {
+                // No callback will run for this submission; handle this path too.
+                std::cerr << "Submission failed for id=" << id << ": " << submitted.error_message
+                          << '\n';
+                break;
+            }
+        }
+        // Submission has stopped. Wait for callbacks before leaving the counters'
+        // scope; individual callback failures are checked below.
+        // check() exits on error; a continuing application must keep callback state alive.
+        // A durable pipeline would advance its source position or offset only
+        // after Flush() succeeds, then replay from there on restart.
+        check("flush", writer.Flush());
+        std::cout << "Callback writes: succeeded=" << succeeded << " failed=" << failed << '\n';
+        if (failed != 0) {
+            return 1;
+        }
+    }
+
     // Append a row with all fields null (matches Rust log_table.rs all_supported_datatypes)
     {
         fluss::GenericRow row;

fluss-rust/bindings/cpp/include/fluss.hpp

@@ -21,6 +21,7 @@
 
 #include <chrono>
 #include <cstdint>
+#include <functional>
 #include <limits>
 #include <memory>
 #include <optional>
@@ -47,6 +48,8 @@ struct Admin;
 struct Table;
 struct AppendWriter;
 struct WriteResult;
+class WriteCallback;
+class WriteCallbackCapacity;
 struct LogScanner;
 struct RecordBatchLogReader;
 struct BatchScanner;
@@ -527,10 +530,58 @@ struct Result {
 
     bool Ok() const { return error_code == 0; }
 
-    /// Returns true if retrying the request may succeed. Client-side errors always return false.
+    /// Returns true if retrying the request may succeed. Does not guarantee that a failed
+    /// write had no effect or that application resubmission is duplicate-safe.
+    /// Client-side errors always return false.
     bool IsRetriable() const { return ErrorCode::IsRetriable(error_code); }
 };
 
+/// Receives the final outcome of an accepted write. Function pointers and lambdas
+/// are supported. An empty callback is rejected before submitting the write.
+///
+/// During normal operation, the SDK owns the callback until completion and invokes
+/// it exactly once on SDK-managed background threads; no caller polling or waiting
+/// thread is needed. Process exit or a crash can prevent delivery. Callbacks never
+/// run inline in the submitting call, but may run concurrently and out of order,
+/// including before the call returns.
+/// Keep callbacks short and synchronize access to shared state, including writers.
+/// Callback overloads do not make writers safe for concurrent access. Captured
+/// references must remain valid until the callback finishes; capturing shared
+/// ownership is recommended. Keep the connection alive until completion.
+///
+/// Success follows the configured acknowledgment policy. Errors are reported after
+/// internal retry handling, but do not guarantee that no data was written.
+/// Application resubmission is a new operation and can produce duplicates even
+/// with SDK idempotence enabled. Retain input identifiers and recovery state as
+/// needed; one callback invocation is not an exactly-once delivery guarantee.
+///
+/// Callback threads are shared across connections. Do not wait for another
+/// callback from a callback: it can exhaust the worker pool. Synchronous SDK
+/// calls require exclusive writer access. Callback submissions to a full writer
+/// fail immediately when called from a callback, instead of blocking the workers.
+/// WriteCallbackOptions bounds outstanding callback operations per writer.
+/// Hand off retries or expensive work without blocking; bound application queues
+/// and handle overflow without silently discarding failed operations.
+///
+/// Exceptions thrown by callbacks are caught and reported to stderr; they do not
+/// change the write outcome or retry the callback. Stop submissions before Flush().
+/// After a successful Rust write flush, Flush() waits up to 60 seconds for pending
+/// callbacks to finish. On error, referenced state may still be in use.
+/// Inside a callback only the callback wait is skipped; the write flush may block.
+/// Flush() does not wait for work handed to application workers or retry queues.
+using WriteCallback = std::function<void(Result)>;
+
+/// Admission limits for callback overloads only; Wait and fire-and-forget are unchanged.
+struct WriteCallbackOptions {
+    /// Maximum operations reserved for submission or awaiting callback completion.
+    /// Must be positive. One AppendArrowBatch call counts as one operation, not its rows.
+    /// Per-writer count, not preallocated storage or a byte limit. Reduce for many
+    /// writers or large captures; independent of Configuration::writer_buffer_memory_size.
+    /// Waiting for a free slot is bounded by client.writer.buffer.wait-timeout, the same
+    /// budget as the buffer-backpressure wait.
+    size_t max_pending_operations = 262144;
+};
+
 struct TablePath {
     std::string database_name;
     std::string table_name;
@@ -1553,9 +1604,12 @@ struct Configuration {
     bool writer_enable_idempotence{true};
     // Maximum number of in-flight requests per bucket for idempotent writes
     size_t writer_max_inflight_requests_per_bucket{5};
-    // Total memory available for buffering write batches (default 64MB)
+    // Shared write-batch memory budget per Connection, across its tables and writers
+    // (default 64 MiB). Not a process RSS limit or a callback-capture memory budget.
     size_t writer_buffer_memory_size{64 * 1024 * 1024};
-    // Maximum time in milliseconds to block waiting for buffer memory
+    // Maximum time in milliseconds to block waiting for buffer memory. This also bounds
+    // the whole callback submission, including callback capacity and buffer backpressure.
+    // UINT64_MAX waits indefinitely; zero fails fast when capacity or memory is unavailable.
     uint64_t writer_buffer_wait_timeout_ms{std::numeric_limits<uint64_t>::max()};
     // Maximum KV backpressure throttle in milliseconds
     uint64_t writer_kv_backpressure_max_throttle_ms{3000};
@@ -1740,6 +1794,8 @@ class TableAppend {
     TableAppend& operator=(TableAppend&&) noexcept = default;
 
     Result CreateWriter(AppendWriter& out);
+    /// Create a writer with per-writer callback admission limits.
+    Result CreateWriter(AppendWriter& out, const WriteCallbackOptions& options);
 
    private:
     friend class Table;
@@ -1759,6 +1815,8 @@ class TableUpsert {
     TableUpsert& PartialUpdateByName(std::vector<std::string> column_names);
 
     Result CreateWriter(UpsertWriter& out);
+    /// Create a writer with per-writer callback admission limits shared by Upsert and Delete.
+    Result CreateWriter(UpsertWriter& out, const WriteCallbackOptions& options);
 
    private:
     friend class Table;
@@ -1877,6 +1935,7 @@ class WriteResult {
     friend class UpsertWriter;
     WriteResult(ffi::WriteResult* inner) noexcept;
 
+    Result Notify(std::unique_ptr<ffi::WriteCallback> callback);
     void Destroy() noexcept;
     ffi::WriteResult* inner_{nullptr};
 };
@@ -1895,17 +1954,37 @@ class AppendWriter {
 
     Result Append(const GenericRow& row);
     Result Append(const GenericRow& row, WriteResult& out);
+    /// Submit a row and notify callback of its final outcome without waiting for
+    /// acknowledgment. Submission is bounded by client.writer.buffer.wait-timeout,
+    /// covering callback capacity and buffer backpressure: within it, Ok means the
+    /// write was accepted and the callback fires exactly once, an error means
+    /// submission failed and no callback runs. A zero timeout makes submission
+    /// non-blocking. See WriteCallbackOptions.
+    Result Append(const GenericRow& row, WriteCallback callback);
     Result AppendArrowBatch(const std::shared_ptr<arrow::RecordBatch>& batch);
     Result AppendArrowBatch(const std::shared_ptr<arrow::RecordBatch>& batch, WriteResult& out);
+    /// Like the callback Append overload, but notifies once for the entire batch.
+    Result AppendArrowBatch(const std::shared_ptr<arrow::RecordBatch>& batch,
+                            WriteCallback callback);
     Result Flush();
 
    private:
     friend class Table;
     friend class TableAppend;
-    AppendWriter(ffi::AppendWriter* writer) noexcept;
+    AppendWriter(ffi::AppendWriter* writer,
+                 std::shared_ptr<ffi::WriteCallbackCapacity> callback_capacity) noexcept;
+
+    // Submit through FFI bounding the buffer-backpressure wait by submit_budget_ms
+    // (Kafka max.block.ms style): negative uses the writer's configured buffer wait
+    // timeout, >= 0 caps the wait at that many ms (0 = fail fast when the buffer is
+    // full). Only the callback path passes a budget; the public overloads pass -1.
+    Result AppendWithBudget(const GenericRow& row, WriteResult& out, int64_t submit_budget_ms);
+    Result AppendArrowBatchWithBudget(const std::shared_ptr<arrow::RecordBatch>& batch,
+                                      WriteResult& out, int64_t submit_budget_ms);
 
     void Destroy() noexcept;
     ffi::AppendWriter* writer_{nullptr};
+    std::shared_ptr<ffi::WriteCallbackCapacity> callback_capacity_;
 };
 
 class UpsertWriter {
@@ -1922,16 +2001,33 @@ class UpsertWriter {
 
     Result Upsert(const GenericRow& row);
     Result Upsert(const GenericRow& row, WriteResult& out);
+    /// Submit an upsert and notify callback of its final outcome without waiting
+    /// for acknowledgment. Submission is bounded by client.writer.buffer.wait-timeout,
+    /// covering callback capacity and buffer backpressure: within it, Ok means the
+    /// write was accepted and the callback fires exactly once, an error means
+    /// submission failed and no callback runs. A zero timeout makes submission
+    /// non-blocking. See WriteCallbackOptions.
+    Result Upsert(const GenericRow& row, WriteCallback callback);
     Result Delete(const GenericRow& row);
     Result Delete(const GenericRow& row, WriteResult& out);
+    /// Like the callback Upsert overload, but deletes a row by primary key.
+    Result Delete(const GenericRow& row, WriteCallback callback);
     Result Flush();
 
    private:
     friend class Table;
     friend class TableUpsert;
-    UpsertWriter(ffi::UpsertWriter* writer) noexcept;
+    UpsertWriter(ffi::UpsertWriter* writer,
+                 std::shared_ptr<ffi::WriteCallbackCapacity> callback_capacity) noexcept;
+
+    // See AppendWriter::AppendWithBudget for submit_budget_ms semantics. Only the
+    // callback path passes a budget; the public overloads pass -1.
+    Result UpsertWithBudget(const GenericRow& row, WriteResult& out, int64_t submit_budget_ms);
+    Result DeleteWithBudget(const GenericRow& row, WriteResult& out, int64_t submit_budget_ms);
+
     void Destroy() noexcept;
     ffi::UpsertWriter* writer_{nullptr};
+    std::shared_ptr<ffi::WriteCallbackCapacity> callback_capacity_;
 };
 
 class Lookuper {