[TASK-389] Idempotent writes

Author: fresh-borzoni

bindings/cpp/include/fluss.hpp

@@ -998,6 +998,14 @@ struct Configuration {
     // Maximum number of records returned in a single call to Poll() for LogScanner
     size_t scanner_log_max_poll_records{500};
     int64_t writer_batch_timeout_ms{100};
+    // Whether to enable idempotent writes
+    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)
+    size_t writer_buffer_memory_size{64 * 1024 * 1024};
+    // Maximum time in milliseconds to block waiting for buffer memory
+    uint64_t writer_buffer_wait_timeout_ms{60000};
     // Connect timeout in milliseconds for TCP transport connect
     uint64_t connect_timeout_ms{120000};
     // Security protocol: "PLAINTEXT" (default, no auth) or "sasl" (SASL auth)

bindings/cpp/src/ffi_converter.hpp

@@ -57,6 +57,11 @@ inline ffi::FfiConfig to_ffi_config(const Configuration& config) {
     ffi_config.scanner_remote_log_read_concurrency = config.scanner_remote_log_read_concurrency;
     ffi_config.scanner_log_max_poll_records = config.scanner_log_max_poll_records;
     ffi_config.writer_batch_timeout_ms = config.writer_batch_timeout_ms;
+    ffi_config.writer_enable_idempotence = config.writer_enable_idempotence;
+    ffi_config.writer_max_inflight_requests_per_bucket =
+        config.writer_max_inflight_requests_per_bucket;
+    ffi_config.writer_buffer_memory_size = config.writer_buffer_memory_size;
+    ffi_config.writer_buffer_wait_timeout_ms = config.writer_buffer_wait_timeout_ms;
     ffi_config.connect_timeout_ms = config.connect_timeout_ms;
     ffi_config.security_protocol = rust::String(config.security_protocol);
     ffi_config.security_sasl_mechanism = rust::String(config.security_sasl_mechanism);

bindings/cpp/src/lib.rs

@@ -49,6 +49,10 @@ mod ffi {
         scanner_remote_log_read_concurrency: usize,
         scanner_log_max_poll_records: usize,
         writer_batch_timeout_ms: i64,
+        writer_enable_idempotence: bool,
+        writer_max_inflight_requests_per_bucket: usize,
+        writer_buffer_memory_size: usize,
+        writer_buffer_wait_timeout_ms: u64,
         connect_timeout_ms: u64,
         security_protocol: String,
         security_sasl_mechanism: String,
@@ -653,6 +657,10 @@ fn new_connection(config: &ffi::FfiConfig) -> Result<*mut Connection, String> {
         remote_file_download_thread_num: config.remote_file_download_thread_num,
         scanner_remote_log_read_concurrency: config.scanner_remote_log_read_concurrency,
         scanner_log_max_poll_records: config.scanner_log_max_poll_records,
+        writer_enable_idempotence: config.writer_enable_idempotence,
+        writer_max_inflight_requests_per_bucket: config.writer_max_inflight_requests_per_bucket,
+        writer_buffer_memory_size: config.writer_buffer_memory_size,
+        writer_buffer_wait_timeout_ms: config.writer_buffer_wait_timeout_ms,
         connect_timeout_ms: config.connect_timeout_ms,
         security_protocol: config.security_protocol.to_string(),
         security_sasl_mechanism: config.security_sasl_mechanism.to_string(),

bindings/python/fluss/__init__.pyi

@@ -174,6 +174,22 @@ class Config:
     @writer_batch_timeout_ms.setter
     def writer_batch_timeout_ms(self, timeout: int) -> None: ...
     @property
+    def writer_enable_idempotence(self) -> bool: ...
+    @writer_enable_idempotence.setter
+    def writer_enable_idempotence(self, enabled: bool) -> None: ...
+    @property
+    def writer_max_inflight_requests_per_bucket(self) -> int: ...
+    @writer_max_inflight_requests_per_bucket.setter
+    def writer_max_inflight_requests_per_bucket(self, num: int) -> None: ...
+    @property
+    def writer_buffer_memory_size(self) -> int: ...
+    @writer_buffer_memory_size.setter
+    def writer_buffer_memory_size(self, size: int) -> None: ...
+    @property
+    def writer_buffer_wait_timeout_ms(self) -> int: ...
+    @writer_buffer_wait_timeout_ms.setter
+    def writer_buffer_wait_timeout_ms(self, timeout: int) -> None: ...
+    @property
     def connect_timeout_ms(self) -> int: ...
     @connect_timeout_ms.setter
     def connect_timeout_ms(self, timeout: int) -> None: ...

bindings/python/src/config.rs

@@ -97,6 +97,38 @@ impl Config {
                                 ))
                             })?;
                     }
+                    "writer.enable-idempotence" => {
+                        config.writer_enable_idempotence = match value.as_str() {
+                            "true" => true,
+                            "false" => false,
+                            other => {
+                                return Err(FlussError::new_err(format!(
+                                    "Invalid value '{other}' for '{key}', expected 'true' or 'false'"
+                                )));
+                            }
+                        };
+                    }
+                    "writer.max-inflight-requests-per-bucket" => {
+                        config.writer_max_inflight_requests_per_bucket =
+                            value.parse::<usize>().map_err(|e| {
+                                FlussError::new_err(format!(
+                                    "Invalid value '{value}' for '{key}': {e}"
+                                ))
+                            })?;
+                    }
+                    "writer.buffer.memory-size" => {
+                        config.writer_buffer_memory_size = value.parse::<usize>().map_err(|e| {
+                            FlussError::new_err(format!("Invalid value '{value}' for '{key}': {e}"))
+                        })?;
+                    }
+                    "writer.buffer.wait-timeout-ms" => {
+                        config.writer_buffer_wait_timeout_ms =
+                            value.parse::<u64>().map_err(|e| {
+                                FlussError::new_err(format!(
+                                    "Invalid value '{value}' for '{key}': {e}"
+                                ))
+                            })?;
+                    }
                     "writer.bucket.no-key-assigner" => {
                         config.writer_bucket_no_key_assigner = match value.as_str() {
                             "round_robin" => fcore::config::NoKeyAssigner::RoundRobin,
@@ -255,6 +287,54 @@ impl Config {
         self.inner.writer_batch_timeout_ms = timeout;
     }
 
+    /// Get whether idempotent writes are enabled
+    #[getter]
+    fn writer_enable_idempotence(&self) -> bool {
+        self.inner.writer_enable_idempotence
+    }
+
+    /// Set whether idempotent writes are enabled
+    #[setter]
+    fn set_writer_enable_idempotence(&mut self, enabled: bool) {
+        self.inner.writer_enable_idempotence = enabled;
+    }
+
+    /// Get the max in-flight requests per bucket
+    #[getter]
+    fn writer_max_inflight_requests_per_bucket(&self) -> usize {
+        self.inner.writer_max_inflight_requests_per_bucket
+    }
+
+    /// Set the max in-flight requests per bucket
+    #[setter]
+    fn set_writer_max_inflight_requests_per_bucket(&mut self, num: usize) {
+        self.inner.writer_max_inflight_requests_per_bucket = num;
+    }
+
+    /// Get the writer buffer memory size
+    #[getter]
+    fn writer_buffer_memory_size(&self) -> usize {
+        self.inner.writer_buffer_memory_size
+    }
+
+    /// Set the writer buffer memory size
+    #[setter]
+    fn set_writer_buffer_memory_size(&mut self, size: usize) {
+        self.inner.writer_buffer_memory_size = size;
+    }
+
+    /// Get the writer buffer wait timeout in milliseconds
+    #[getter]
+    fn writer_buffer_wait_timeout_ms(&self) -> u64 {
+        self.inner.writer_buffer_wait_timeout_ms
+    }
+
+    /// Set the writer buffer wait timeout in milliseconds
+    #[setter]
+    fn set_writer_buffer_wait_timeout_ms(&mut self, timeout: u64) {
+        self.inner.writer_buffer_wait_timeout_ms = timeout;
+    }
+
     /// Get the connect timeout in milliseconds
     #[getter]
     fn connect_timeout_ms(&self) -> u64 {