@@ -97,6 +97,64 @@ impl Config {
9797 ) )
9898 } ) ?;
9999 }
100+ "scanner.log.fetch.max-bytes" => {
101+ config. scanner_log_fetch_max_bytes = value. parse :: < i32 > ( ) . map_err ( |e| {
102+ FlussError :: new_err ( format ! ( "Invalid value '{value}' for '{key}': {e}" ) )
103+ } ) ?;
104+ }
105+ "scanner.log.fetch.min-bytes" => {
106+ config. scanner_log_fetch_min_bytes = value. parse :: < i32 > ( ) . map_err ( |e| {
107+ FlussError :: new_err ( format ! ( "Invalid value '{value}' for '{key}': {e}" ) )
108+ } ) ?;
109+ }
110+ "scanner.log.fetch.wait-max-time-ms" => {
111+ config. scanner_log_fetch_wait_max_time_ms =
112+ value. parse :: < i32 > ( ) . map_err ( |e| {
113+ FlussError :: new_err ( format ! (
114+ "Invalid value '{value}' for '{key}': {e}"
115+ ) )
116+ } ) ?;
117+ }
118+ "scanner.log.fetch.max-bytes-for-bucket" => {
119+ config. scanner_log_fetch_max_bytes_for_bucket =
120+ value. parse :: < i32 > ( ) . map_err ( |e| {
121+ FlussError :: new_err ( format ! (
122+ "Invalid value '{value}' for '{key}': {e}"
123+ ) )
124+ } ) ?;
125+ }
126+ "writer.enable-idempotence" => {
127+ config. writer_enable_idempotence = match value. as_str ( ) {
128+ "true" => true ,
129+ "false" => false ,
130+ other => {
131+ return Err ( FlussError :: new_err ( format ! (
132+ "Invalid value '{other}' for '{key}', expected 'true' or 'false'"
133+ ) ) ) ;
134+ }
135+ } ;
136+ }
137+ "writer.max-inflight-requests-per-bucket" => {
138+ config. writer_max_inflight_requests_per_bucket =
139+ value. parse :: < usize > ( ) . map_err ( |e| {
140+ FlussError :: new_err ( format ! (
141+ "Invalid value '{value}' for '{key}': {e}"
142+ ) )
143+ } ) ?;
144+ }
145+ "writer.buffer.memory-size" => {
146+ config. writer_buffer_memory_size = value. parse :: < usize > ( ) . map_err ( |e| {
147+ FlussError :: new_err ( format ! ( "Invalid value '{value}' for '{key}': {e}" ) )
148+ } ) ?;
149+ }
150+ "writer.buffer.wait-timeout-ms" => {
151+ config. writer_buffer_wait_timeout_ms =
152+ value. parse :: < u64 > ( ) . map_err ( |e| {
153+ FlussError :: new_err ( format ! (
154+ "Invalid value '{value}' for '{key}': {e}"
155+ ) )
156+ } ) ?;
157+ }
100158 "writer.bucket.no-key-assigner" => {
101159 config. writer_bucket_no_key_assigner =
102160 value. parse :: < fcore:: config:: NoKeyAssigner > ( ) . map_err ( |e| {
@@ -270,6 +328,54 @@ impl Config {
270328 Ok ( ( ) )
271329 }
272330
331+ /// Get whether idempotent writes are enabled
332+ #[ getter]
333+ fn writer_enable_idempotence ( & self ) -> bool {
334+ self . inner . writer_enable_idempotence
335+ }
336+
337+ /// Set whether idempotent writes are enabled
338+ #[ setter]
339+ fn set_writer_enable_idempotence ( & mut self , enabled : bool ) {
340+ self . inner . writer_enable_idempotence = enabled;
341+ }
342+
343+ /// Get the max in-flight requests per bucket
344+ #[ getter]
345+ fn writer_max_inflight_requests_per_bucket ( & self ) -> usize {
346+ self . inner . writer_max_inflight_requests_per_bucket
347+ }
348+
349+ /// Set the max in-flight requests per bucket
350+ #[ setter]
351+ fn set_writer_max_inflight_requests_per_bucket ( & mut self , num : usize ) {
352+ self . inner . writer_max_inflight_requests_per_bucket = num;
353+ }
354+
355+ /// Get the writer buffer memory size
356+ #[ getter]
357+ fn writer_buffer_memory_size ( & self ) -> usize {
358+ self . inner . writer_buffer_memory_size
359+ }
360+
361+ /// Set the writer buffer memory size
362+ #[ setter]
363+ fn set_writer_buffer_memory_size ( & mut self , size : usize ) {
364+ self . inner . writer_buffer_memory_size = size;
365+ }
366+
367+ /// Get the writer buffer wait timeout in milliseconds
368+ #[ getter]
369+ fn writer_buffer_wait_timeout_ms ( & self ) -> u64 {
370+ self . inner . writer_buffer_wait_timeout_ms
371+ }
372+
373+ /// Set the writer buffer wait timeout in milliseconds
374+ #[ setter]
375+ fn set_writer_buffer_wait_timeout_ms ( & mut self , timeout : u64 ) {
376+ self . inner . writer_buffer_wait_timeout_ms = timeout;
377+ }
378+
273379 /// Get the connect timeout in milliseconds
274380#[ getter]
275381 fn connect_timeout_ms ( & self ) -> u64 {
@@ -329,6 +435,54 @@ impl Config {
329435 fn set_security_sasl_password ( & mut self , password : String ) {
330436 self . inner . security_sasl_password = password;
331437 }
438+
439+ /// Get the maximum bytes per fetch response for LogScanner
440+ #[ getter]
441+ fn scanner_log_fetch_max_bytes ( & self ) -> i32 {
442+ self . inner . scanner_log_fetch_max_bytes
443+ }
444+
445+ /// Set the maximum bytes per fetch response for LogScanner
446+ #[ setter]
447+ fn set_scanner_log_fetch_max_bytes ( & mut self , bytes : i32 ) {
448+ self . inner . scanner_log_fetch_max_bytes = bytes;
449+ }
450+
451+ /// Get the minimum bytes to accumulate before returning a fetch response
452+ #[ getter]
453+ fn scanner_log_fetch_min_bytes ( & self ) -> i32 {
454+ self . inner . scanner_log_fetch_min_bytes
455+ }
456+
457+ /// Set the minimum bytes to accumulate before returning a fetch response
458+ #[ setter]
459+ fn set_scanner_log_fetch_min_bytes ( & mut self , bytes : i32 ) {
460+ self . inner . scanner_log_fetch_min_bytes = bytes;
461+ }
462+
463+ /// Get the maximum time (ms) the server may wait to satisfy min-bytes
464+ #[ getter]
465+ fn scanner_log_fetch_wait_max_time_ms ( & self ) -> i32 {
466+ self . inner . scanner_log_fetch_wait_max_time_ms
467+ }
468+
469+ /// Set the maximum time (ms) the server may wait to satisfy min-bytes
470+ #[ setter]
471+ fn set_scanner_log_fetch_wait_max_time_ms ( & mut self , ms : i32 ) {
472+ self . inner . scanner_log_fetch_wait_max_time_ms = ms;
473+ }
474+
475+ /// Get the maximum bytes per fetch response per bucket for LogScanner
476+ #[ getter]
477+ fn scanner_log_fetch_max_bytes_for_bucket ( & self ) -> i32 {
478+ self . inner . scanner_log_fetch_max_bytes_for_bucket
479+ }
480+
481+ /// Set the maximum bytes per fetch response per bucket for LogScanner
482+ #[ setter]
483+ fn set_scanner_log_fetch_max_bytes_for_bucket ( & mut self , bytes : i32 ) {
484+ self . inner . scanner_log_fetch_max_bytes_for_bucket = bytes;
485+ }
332486}
333487
334488impl Config {
0 commit comments