You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "beryllw (via GitHub)" <gi...@apache.org> on 2026/08/22 03:38:29 UTC
[PR] [rust] Add atomic property overlays for Config [fluss]
beryllw opened a new pull request, #4068:
URL: https://github.com/apache/fluss/pull/4068
<!--
*Thank you very much for contributing to Fluss - we are happy that you want to help us improve Fluss. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [GitHub issue](https://github.com/apache/fluss/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no issue.
- Name the pull request in the format "[component] Title of the pull request", where *[component]* should be replaced by the name of the component being changed. Typically, this corresponds to the component label assigned to the issue (e.g., [kv], [log], [client], [flink], [rust], [python], [c++], [elixir]). Skip *[component]* if you are unsure about which is the best component.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes.
- Each pull request should address only one issue, not mix up code from multiple issues.
- **Generative AI disclosure:** Indicate whether generative AI tools were used in authoring this PR. If yes, specify the tool below.
- [ ] No generative AI tools used
- [ ] Yes (please specify the tool below)
**(The sections below can be removed for hotfixes or typos)**
-->
<!--
Generated-by: [Tool Name and Version] following [the guidelines](https://github.com/apache/fluss/blob/main/AGENTS.md)
-->
### Purpose
<!-- Linking this pull request to the issue -->
Linked issue: close #4064
<!-- What is the purpose of the change -->
### Brief change log
<!-- Please describe the changes made in this pull request and explain how they address the issue -->
### Tests
<!-- List UT and IT cases to verify this change -->
### API and Format
<!-- Does this change affect API or storage format -->
### Documentation
<!-- Does this change introduce a new feature -->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rust] Add property overlays for rust client Config [fluss]
Posted by "beryllw (via GitHub)" <gi...@apache.org>.
beryllw commented on PR #4068:
URL: https://github.com/apache/fluss/pull/4068#issuecomment-5383873580
@fresh-borzoni @luoyuxia @loserwang1024 cc
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rust] Add property overlays for rust client Config [fluss]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #4068:
URL: https://github.com/apache/fluss/pull/4068#discussion_r3852756729
##########
fluss-rust/crates/fluss/src/config.rs:
##########
@@ -212,6 +213,250 @@ pub struct Config {
pub lookup_max_retries: i32,
}
+type ConfigPropertyApplier = fn(&mut Config, &str) -> Result<(), String>;
+
+struct ConfigPropertySpec {
+ key: &'static str,
+ sensitive: bool,
+ apply: ConfigPropertyApplier,
+}
+
+const CONFIG_PROPERTY_SPECS: &[ConfigPropertySpec] = &[
+ ConfigPropertySpec {
+ key: "security.protocol",
+ sensitive: false,
+ apply: |config, value| {
+ config.security_protocol = value.to_string();
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "security.sasl.mechanism",
+ sensitive: false,
+ apply: |config, value| {
+ config.security_sasl_mechanism = value.to_string();
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "security.sasl.username",
+ sensitive: false,
+ apply: |config, value| {
+ config.security_sasl_username = value.to_string();
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "security.sasl.password",
+ sensitive: true,
+ apply: |config, value| {
+ config.security_sasl_password = value.to_string();
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "connect-timeout",
Review Comment:
Could we freeze the compatibility contract before exposing this API? Python currently accepts bare-millisecond values and keys such as `writer.batch-timeout-ms`, while this overlay requires units and different dotted keys. Please add aliases (or a documented migration) plus cross-binding compatibility tests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rust] Add property overlays for rust client Config [fluss]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #4068:
URL: https://github.com/apache/fluss/pull/4068#discussion_r3852753244
##########
fluss-rust/crates/fluss/src/config.rs:
##########
@@ -212,6 +213,250 @@ pub struct Config {
pub lookup_max_retries: i32,
}
+type ConfigPropertyApplier = fn(&mut Config, &str) -> Result<(), String>;
+
+struct ConfigPropertySpec {
+ key: &'static str,
+ sensitive: bool,
+ apply: ConfigPropertyApplier,
+}
+
+const CONFIG_PROPERTY_SPECS: &[ConfigPropertySpec] = &[
+ ConfigPropertySpec {
+ key: "security.protocol",
+ sensitive: false,
+ apply: |config, value| {
+ config.security_protocol = value.to_string();
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "security.sasl.mechanism",
+ sensitive: false,
+ apply: |config, value| {
+ config.security_sasl_mechanism = value.to_string();
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "security.sasl.username",
+ sensitive: false,
+ apply: |config, value| {
+ config.security_sasl_username = value.to_string();
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "security.sasl.password",
+ sensitive: true,
+ apply: |config, value| {
+ config.security_sasl_password = value.to_string();
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "connect-timeout",
+ sensitive: false,
+ apply: |config, value| {
+ config.connect_timeout_ms = parse_duration_ms(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.batch-size",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_batch_size = parse_i32_bytes(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.request-max-size",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_request_max_size = parse_i32_bytes(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.buffer.memory-size",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_buffer_memory_size = parse_usize_bytes(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.buffer.wait-timeout",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_buffer_wait_timeout_ms = parse_duration_ms(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.batch-timeout",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_batch_timeout_ms = i64::try_from(parse_duration_ms(value)?)
+ .map_err(|_| "duration exceeds the supported range".to_string())?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.dynamic-batch-size.enabled",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_dynamic_batch_size_enabled = parse_bool(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.dynamic-batch-size.min",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_dynamic_batch_size_min = parse_i32_bytes(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.kv-backpressure.max-throttle",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_kv_backpressure_max_throttle_ms = parse_duration_ms(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "lookup.queue-size",
+ sensitive: false,
+ apply: |config, value| {
+ config.lookup_queue_size = parse_usize_count(value, 1)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "lookup.max-batch-size",
+ sensitive: false,
+ apply: |config, value| {
+ config.lookup_max_batch_size = parse_usize_count(value, 1)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "lookup.max-inflight-requests",
+ sensitive: false,
+ apply: |config, value| {
+ config.lookup_max_inflight_requests = parse_usize_count(value, 1)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "lookup.max-retries",
+ sensitive: false,
+ apply: |config, value| {
+ config.lookup_max_retries = parse_i32_count(value, 0)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "lookup.batch-timeout",
+ sensitive: false,
+ apply: |config, value| {
+ config.lookup_batch_timeout_ms = parse_duration_ms(value)?;
+ Ok(())
+ },
+ },
+];
+
+fn config_property_spec(key: &str) -> Option<&'static ConfigPropertySpec> {
+ CONFIG_PROPERTY_SPECS.iter().find(|spec| spec.key == key)
+}
+
+fn parse_bool(value: &str) -> Result<bool, String> {
+ value
+ .parse::<bool>()
+ .map_err(|_| "expected 'true' or 'false'".to_string())
+}
+
+fn parse_i32_count(value: &str, min: i32) -> Result<i32, String> {
+ let parsed = value
+ .parse::<i32>()
+ .map_err(|_| "expected a 32-bit integer".to_string())?;
+ if parsed < min {
+ return Err(format!("must be at least {min}"));
+ }
+ Ok(parsed)
+}
+
+fn parse_usize_count(value: &str, min: usize) -> Result<usize, String> {
+ let parsed = value
+ .parse::<usize>()
+ .map_err(|_| "expected a non-negative integer".to_string())?;
+ if parsed < min {
+ return Err(format!("must be at least {min}"));
+ }
+ Ok(parsed)
+}
+
+fn parse_i32_bytes(value: &str) -> Result<i32, String> {
+ i32::try_from(parse_byte_size(value)?)
+ .map_err(|_| "byte size exceeds the supported range".to_string())
+}
+
+fn parse_usize_bytes(value: &str) -> Result<usize, String> {
+ usize::try_from(parse_byte_size(value)?)
+ .map_err(|_| "byte size exceeds the supported range".to_string())
+}
+
+fn parse_byte_size(value: &str) -> Result<u64, String> {
+ let (number, unit) = split_number_and_unit(value)?;
+ let multiplier = match unit.to_ascii_lowercase().as_str() {
+ "" | "b" | "bytes" => 1,
+ "k" | "kb" | "kib" | "kibibyte" | "kibibytes" => 1_024,
+ "m" | "mb" | "mib" | "mebibyte" | "mebibytes" => 1_024 * 1_024,
+ "g" | "gb" | "gib" | "gibibyte" | "gibibytes" => 1_024 * 1_024 * 1_024,
+ "t" | "tb" | "tib" | "tebibyte" | "tebibytes" => 1_024_u64 * 1_024 * 1_024 * 1_024,
+ _ => return Err("unsupported byte-size unit".to_string()),
+ };
+ let bytes = number
+ .checked_mul(multiplier)
+ .ok_or_else(|| "byte size exceeds the supported range".to_string())?;
+ if bytes == 0 {
+ return Err("byte size must be greater than zero".to_string());
+ }
+ Ok(bytes)
+}
+
+fn parse_duration_ms(value: &str) -> Result<u64, String> {
Review Comment:
Could zero be validated per property rather than by the generic duration parser? `lookup.batch-timeout=0ms` currently passes validation; `LookupQueue::drain()` then exits before `recv()`, leaving requests queued while `LookupSender` can spin on empty batches. Please reject zero for connect/lookup timeouts and add a runtime regression test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rust] Add property overlays for rust client Config [fluss]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #4068:
URL: https://github.com/apache/fluss/pull/4068#discussion_r3852753244
##########
fluss-rust/crates/fluss/src/config.rs:
##########
@@ -212,6 +213,250 @@ pub struct Config {
pub lookup_max_retries: i32,
}
+type ConfigPropertyApplier = fn(&mut Config, &str) -> Result<(), String>;
+
+struct ConfigPropertySpec {
+ key: &'static str,
+ sensitive: bool,
+ apply: ConfigPropertyApplier,
+}
+
+const CONFIG_PROPERTY_SPECS: &[ConfigPropertySpec] = &[
+ ConfigPropertySpec {
+ key: "security.protocol",
+ sensitive: false,
+ apply: |config, value| {
+ config.security_protocol = value.to_string();
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "security.sasl.mechanism",
+ sensitive: false,
+ apply: |config, value| {
+ config.security_sasl_mechanism = value.to_string();
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "security.sasl.username",
+ sensitive: false,
+ apply: |config, value| {
+ config.security_sasl_username = value.to_string();
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "security.sasl.password",
+ sensitive: true,
+ apply: |config, value| {
+ config.security_sasl_password = value.to_string();
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "connect-timeout",
+ sensitive: false,
+ apply: |config, value| {
+ config.connect_timeout_ms = parse_duration_ms(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.batch-size",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_batch_size = parse_i32_bytes(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.request-max-size",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_request_max_size = parse_i32_bytes(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.buffer.memory-size",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_buffer_memory_size = parse_usize_bytes(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.buffer.wait-timeout",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_buffer_wait_timeout_ms = parse_duration_ms(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.batch-timeout",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_batch_timeout_ms = i64::try_from(parse_duration_ms(value)?)
+ .map_err(|_| "duration exceeds the supported range".to_string())?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.dynamic-batch-size.enabled",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_dynamic_batch_size_enabled = parse_bool(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.dynamic-batch-size.min",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_dynamic_batch_size_min = parse_i32_bytes(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.kv-backpressure.max-throttle",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_kv_backpressure_max_throttle_ms = parse_duration_ms(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "lookup.queue-size",
+ sensitive: false,
+ apply: |config, value| {
+ config.lookup_queue_size = parse_usize_count(value, 1)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "lookup.max-batch-size",
+ sensitive: false,
+ apply: |config, value| {
+ config.lookup_max_batch_size = parse_usize_count(value, 1)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "lookup.max-inflight-requests",
+ sensitive: false,
+ apply: |config, value| {
+ config.lookup_max_inflight_requests = parse_usize_count(value, 1)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "lookup.max-retries",
+ sensitive: false,
+ apply: |config, value| {
+ config.lookup_max_retries = parse_i32_count(value, 0)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "lookup.batch-timeout",
+ sensitive: false,
+ apply: |config, value| {
+ config.lookup_batch_timeout_ms = parse_duration_ms(value)?;
+ Ok(())
+ },
+ },
+];
+
+fn config_property_spec(key: &str) -> Option<&'static ConfigPropertySpec> {
+ CONFIG_PROPERTY_SPECS.iter().find(|spec| spec.key == key)
+}
+
+fn parse_bool(value: &str) -> Result<bool, String> {
+ value
+ .parse::<bool>()
+ .map_err(|_| "expected 'true' or 'false'".to_string())
+}
+
+fn parse_i32_count(value: &str, min: i32) -> Result<i32, String> {
+ let parsed = value
+ .parse::<i32>()
+ .map_err(|_| "expected a 32-bit integer".to_string())?;
+ if parsed < min {
+ return Err(format!("must be at least {min}"));
+ }
+ Ok(parsed)
+}
+
+fn parse_usize_count(value: &str, min: usize) -> Result<usize, String> {
+ let parsed = value
+ .parse::<usize>()
+ .map_err(|_| "expected a non-negative integer".to_string())?;
+ if parsed < min {
+ return Err(format!("must be at least {min}"));
+ }
+ Ok(parsed)
+}
+
+fn parse_i32_bytes(value: &str) -> Result<i32, String> {
+ i32::try_from(parse_byte_size(value)?)
+ .map_err(|_| "byte size exceeds the supported range".to_string())
+}
+
+fn parse_usize_bytes(value: &str) -> Result<usize, String> {
+ usize::try_from(parse_byte_size(value)?)
+ .map_err(|_| "byte size exceeds the supported range".to_string())
+}
+
+fn parse_byte_size(value: &str) -> Result<u64, String> {
+ let (number, unit) = split_number_and_unit(value)?;
+ let multiplier = match unit.to_ascii_lowercase().as_str() {
+ "" | "b" | "bytes" => 1,
+ "k" | "kb" | "kib" | "kibibyte" | "kibibytes" => 1_024,
+ "m" | "mb" | "mib" | "mebibyte" | "mebibytes" => 1_024 * 1_024,
+ "g" | "gb" | "gib" | "gibibyte" | "gibibytes" => 1_024 * 1_024 * 1_024,
+ "t" | "tb" | "tib" | "tebibyte" | "tebibytes" => 1_024_u64 * 1_024 * 1_024 * 1_024,
+ _ => return Err("unsupported byte-size unit".to_string()),
+ };
+ let bytes = number
+ .checked_mul(multiplier)
+ .ok_or_else(|| "byte size exceeds the supported range".to_string())?;
+ if bytes == 0 {
+ return Err("byte size must be greater than zero".to_string());
+ }
+ Ok(bytes)
+}
+
+fn parse_duration_ms(value: &str) -> Result<u64, String> {
Review Comment:
Could zero be validated per property rather than by the generic duration parser? `lookup.batch-timeout=0ms` currently passes validation; `LookupQueue::drain()` then exits before `recv()`, leaving requests queued while `LookupSender` can spin on empty batches. Please reject zero for connect/lookup timeouts and add a runtime regression test.
##########
fluss-rust/crates/fluss/src/config.rs:
##########
@@ -212,6 +213,250 @@ pub struct Config {
pub lookup_max_retries: i32,
}
+type ConfigPropertyApplier = fn(&mut Config, &str) -> Result<(), String>;
+
+struct ConfigPropertySpec {
+ key: &'static str,
+ sensitive: bool,
+ apply: ConfigPropertyApplier,
+}
+
+const CONFIG_PROPERTY_SPECS: &[ConfigPropertySpec] = &[
+ ConfigPropertySpec {
+ key: "security.protocol",
+ sensitive: false,
+ apply: |config, value| {
+ config.security_protocol = value.to_string();
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "security.sasl.mechanism",
+ sensitive: false,
+ apply: |config, value| {
+ config.security_sasl_mechanism = value.to_string();
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "security.sasl.username",
+ sensitive: false,
+ apply: |config, value| {
+ config.security_sasl_username = value.to_string();
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "security.sasl.password",
+ sensitive: true,
+ apply: |config, value| {
+ config.security_sasl_password = value.to_string();
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "connect-timeout",
Review Comment:
Could we freeze the compatibility contract before exposing this API? Python currently accepts bare-millisecond values and keys such as `writer.batch-timeout-ms`, while this overlay requires units and different dotted keys. Please add aliases (or a documented migration) plus cross-binding compatibility tests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rust] Add property overlays for rust client Config [fluss]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #4068:
URL: https://github.com/apache/fluss/pull/4068#discussion_r3852745750
##########
fluss-rust/crates/fluss/src/config.rs:
##########
@@ -212,6 +213,250 @@ pub struct Config {
pub lookup_max_retries: i32,
}
+type ConfigPropertyApplier = fn(&mut Config, &str) -> Result<(), String>;
+
+struct ConfigPropertySpec {
+ key: &'static str,
+ sensitive: bool,
+ apply: ConfigPropertyApplier,
+}
+
+const CONFIG_PROPERTY_SPECS: &[ConfigPropertySpec] = &[
Review Comment:
Could we either cover the full `Config` property surface (especially `bootstrap.servers` and `scanner.*`) or explicitly document this as a supported subset? As written, common configurations cannot be expressed through the overlay, and no production caller exercises the contract.
##########
fluss-rust/crates/fluss/src/config.rs:
##########
@@ -212,6 +213,250 @@ pub struct Config {
pub lookup_max_retries: i32,
}
+type ConfigPropertyApplier = fn(&mut Config, &str) -> Result<(), String>;
+
+struct ConfigPropertySpec {
+ key: &'static str,
+ sensitive: bool,
+ apply: ConfigPropertyApplier,
+}
+
+const CONFIG_PROPERTY_SPECS: &[ConfigPropertySpec] = &[
+ ConfigPropertySpec {
+ key: "security.protocol",
+ sensitive: false,
+ apply: |config, value| {
+ config.security_protocol = value.to_string();
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "security.sasl.mechanism",
+ sensitive: false,
+ apply: |config, value| {
+ config.security_sasl_mechanism = value.to_string();
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "security.sasl.username",
+ sensitive: false,
+ apply: |config, value| {
+ config.security_sasl_username = value.to_string();
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "security.sasl.password",
+ sensitive: true,
+ apply: |config, value| {
+ config.security_sasl_password = value.to_string();
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "connect-timeout",
+ sensitive: false,
+ apply: |config, value| {
+ config.connect_timeout_ms = parse_duration_ms(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.batch-size",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_batch_size = parse_i32_bytes(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.request-max-size",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_request_max_size = parse_i32_bytes(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.buffer.memory-size",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_buffer_memory_size = parse_usize_bytes(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.buffer.wait-timeout",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_buffer_wait_timeout_ms = parse_duration_ms(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.batch-timeout",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_batch_timeout_ms = i64::try_from(parse_duration_ms(value)?)
+ .map_err(|_| "duration exceeds the supported range".to_string())?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.dynamic-batch-size.enabled",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_dynamic_batch_size_enabled = parse_bool(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.dynamic-batch-size.min",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_dynamic_batch_size_min = parse_i32_bytes(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.kv-backpressure.max-throttle",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_kv_backpressure_max_throttle_ms = parse_duration_ms(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "lookup.queue-size",
+ sensitive: false,
+ apply: |config, value| {
+ config.lookup_queue_size = parse_usize_count(value, 1)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "lookup.max-batch-size",
+ sensitive: false,
+ apply: |config, value| {
+ config.lookup_max_batch_size = parse_usize_count(value, 1)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "lookup.max-inflight-requests",
+ sensitive: false,
+ apply: |config, value| {
+ config.lookup_max_inflight_requests = parse_usize_count(value, 1)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "lookup.max-retries",
+ sensitive: false,
+ apply: |config, value| {
+ config.lookup_max_retries = parse_i32_count(value, 0)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "lookup.batch-timeout",
+ sensitive: false,
+ apply: |config, value| {
+ config.lookup_batch_timeout_ms = parse_duration_ms(value)?;
+ Ok(())
+ },
+ },
+];
+
+fn config_property_spec(key: &str) -> Option<&'static ConfigPropertySpec> {
+ CONFIG_PROPERTY_SPECS.iter().find(|spec| spec.key == key)
+}
+
+fn parse_bool(value: &str) -> Result<bool, String> {
+ value
+ .parse::<bool>()
+ .map_err(|_| "expected 'true' or 'false'".to_string())
+}
+
+fn parse_i32_count(value: &str, min: i32) -> Result<i32, String> {
+ let parsed = value
+ .parse::<i32>()
+ .map_err(|_| "expected a 32-bit integer".to_string())?;
+ if parsed < min {
+ return Err(format!("must be at least {min}"));
+ }
+ Ok(parsed)
+}
+
+fn parse_usize_count(value: &str, min: usize) -> Result<usize, String> {
+ let parsed = value
+ .parse::<usize>()
+ .map_err(|_| "expected a non-negative integer".to_string())?;
+ if parsed < min {
+ return Err(format!("must be at least {min}"));
+ }
+ Ok(parsed)
+}
+
+fn parse_i32_bytes(value: &str) -> Result<i32, String> {
+ i32::try_from(parse_byte_size(value)?)
+ .map_err(|_| "byte size exceeds the supported range".to_string())
+}
+
+fn parse_usize_bytes(value: &str) -> Result<usize, String> {
+ usize::try_from(parse_byte_size(value)?)
+ .map_err(|_| "byte size exceeds the supported range".to_string())
+}
+
+fn parse_byte_size(value: &str) -> Result<u64, String> {
Review Comment:
Could we define one canonical parser specifically for client properties instead of duplicating the gateway parser? The semantics already differ (`KB` is 1000 in the gateway but 1024 here), so directly sharing either implementation would change behavior. The client-property parser should own its units and aliases.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rust] Add property overlays for rust client Config [fluss]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #4068:
URL: https://github.com/apache/fluss/pull/4068#discussion_r3852750095
##########
fluss-rust/crates/fluss/src/config.rs:
##########
@@ -212,6 +213,250 @@ pub struct Config {
pub lookup_max_retries: i32,
}
+type ConfigPropertyApplier = fn(&mut Config, &str) -> Result<(), String>;
+
+struct ConfigPropertySpec {
+ key: &'static str,
+ sensitive: bool,
+ apply: ConfigPropertyApplier,
+}
+
+const CONFIG_PROPERTY_SPECS: &[ConfigPropertySpec] = &[
+ ConfigPropertySpec {
+ key: "security.protocol",
+ sensitive: false,
+ apply: |config, value| {
+ config.security_protocol = value.to_string();
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "security.sasl.mechanism",
+ sensitive: false,
+ apply: |config, value| {
+ config.security_sasl_mechanism = value.to_string();
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "security.sasl.username",
+ sensitive: false,
+ apply: |config, value| {
+ config.security_sasl_username = value.to_string();
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "security.sasl.password",
+ sensitive: true,
+ apply: |config, value| {
+ config.security_sasl_password = value.to_string();
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "connect-timeout",
+ sensitive: false,
+ apply: |config, value| {
+ config.connect_timeout_ms = parse_duration_ms(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.batch-size",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_batch_size = parse_i32_bytes(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.request-max-size",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_request_max_size = parse_i32_bytes(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.buffer.memory-size",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_buffer_memory_size = parse_usize_bytes(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.buffer.wait-timeout",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_buffer_wait_timeout_ms = parse_duration_ms(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.batch-timeout",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_batch_timeout_ms = i64::try_from(parse_duration_ms(value)?)
+ .map_err(|_| "duration exceeds the supported range".to_string())?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.dynamic-batch-size.enabled",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_dynamic_batch_size_enabled = parse_bool(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.dynamic-batch-size.min",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_dynamic_batch_size_min = parse_i32_bytes(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.kv-backpressure.max-throttle",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_kv_backpressure_max_throttle_ms = parse_duration_ms(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "lookup.queue-size",
+ sensitive: false,
+ apply: |config, value| {
+ config.lookup_queue_size = parse_usize_count(value, 1)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "lookup.max-batch-size",
+ sensitive: false,
+ apply: |config, value| {
+ config.lookup_max_batch_size = parse_usize_count(value, 1)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "lookup.max-inflight-requests",
+ sensitive: false,
+ apply: |config, value| {
+ config.lookup_max_inflight_requests = parse_usize_count(value, 1)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "lookup.max-retries",
+ sensitive: false,
+ apply: |config, value| {
+ config.lookup_max_retries = parse_i32_count(value, 0)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "lookup.batch-timeout",
+ sensitive: false,
+ apply: |config, value| {
+ config.lookup_batch_timeout_ms = parse_duration_ms(value)?;
+ Ok(())
+ },
+ },
+];
+
+fn config_property_spec(key: &str) -> Option<&'static ConfigPropertySpec> {
+ CONFIG_PROPERTY_SPECS.iter().find(|spec| spec.key == key)
+}
+
+fn parse_bool(value: &str) -> Result<bool, String> {
+ value
+ .parse::<bool>()
+ .map_err(|_| "expected 'true' or 'false'".to_string())
+}
+
+fn parse_i32_count(value: &str, min: i32) -> Result<i32, String> {
+ let parsed = value
+ .parse::<i32>()
+ .map_err(|_| "expected a 32-bit integer".to_string())?;
+ if parsed < min {
+ return Err(format!("must be at least {min}"));
+ }
+ Ok(parsed)
+}
+
+fn parse_usize_count(value: &str, min: usize) -> Result<usize, String> {
+ let parsed = value
+ .parse::<usize>()
+ .map_err(|_| "expected a non-negative integer".to_string())?;
+ if parsed < min {
+ return Err(format!("must be at least {min}"));
+ }
+ Ok(parsed)
+}
+
+fn parse_i32_bytes(value: &str) -> Result<i32, String> {
+ i32::try_from(parse_byte_size(value)?)
+ .map_err(|_| "byte size exceeds the supported range".to_string())
+}
+
+fn parse_usize_bytes(value: &str) -> Result<usize, String> {
+ usize::try_from(parse_byte_size(value)?)
+ .map_err(|_| "byte size exceeds the supported range".to_string())
+}
+
+fn parse_byte_size(value: &str) -> Result<u64, String> {
Review Comment:
Could we define one canonical parser specifically for client properties instead of duplicating the gateway parser? The semantics already differ (`KB` is 1000 in the gateway but 1024 here), so directly sharing either implementation would change behavior. The client-property parser should own its units and aliases.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rust] Add property overlays for rust client Config [fluss]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #4068:
URL: https://github.com/apache/fluss/pull/4068#discussion_r3852745750
##########
fluss-rust/crates/fluss/src/config.rs:
##########
@@ -212,6 +213,250 @@ pub struct Config {
pub lookup_max_retries: i32,
}
+type ConfigPropertyApplier = fn(&mut Config, &str) -> Result<(), String>;
+
+struct ConfigPropertySpec {
+ key: &'static str,
+ sensitive: bool,
+ apply: ConfigPropertyApplier,
+}
+
+const CONFIG_PROPERTY_SPECS: &[ConfigPropertySpec] = &[
Review Comment:
Could we either cover the full `Config` property surface (especially `bootstrap.servers` and `scanner.*`) or explicitly document this as a supported subset? As written, common configurations cannot be expressed through the overlay, and no production caller exercises the contract.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rust] Add property overlays for rust client Config [fluss]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #4068:
URL: https://github.com/apache/fluss/pull/4068#discussion_r3852901316
##########
fluss-rust/crates/fluss/src/config.rs:
##########
@@ -212,6 +213,250 @@ pub struct Config {
pub lookup_max_retries: i32,
}
+type ConfigPropertyApplier = fn(&mut Config, &str) -> Result<(), String>;
+
+struct ConfigPropertySpec {
+ key: &'static str,
+ sensitive: bool,
+ apply: ConfigPropertyApplier,
+}
+
+const CONFIG_PROPERTY_SPECS: &[ConfigPropertySpec] = &[
+ ConfigPropertySpec {
+ key: "security.protocol",
+ sensitive: false,
+ apply: |config, value| {
+ config.security_protocol = value.to_string();
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "security.sasl.mechanism",
+ sensitive: false,
+ apply: |config, value| {
+ config.security_sasl_mechanism = value.to_string();
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "security.sasl.username",
+ sensitive: false,
+ apply: |config, value| {
+ config.security_sasl_username = value.to_string();
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "security.sasl.password",
+ sensitive: true,
+ apply: |config, value| {
+ config.security_sasl_password = value.to_string();
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "connect-timeout",
+ sensitive: false,
+ apply: |config, value| {
+ config.connect_timeout_ms = parse_duration_ms(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.batch-size",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_batch_size = parse_i32_bytes(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.request-max-size",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_request_max_size = parse_i32_bytes(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.buffer.memory-size",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_buffer_memory_size = parse_usize_bytes(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.buffer.wait-timeout",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_buffer_wait_timeout_ms = parse_duration_ms(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.batch-timeout",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_batch_timeout_ms = i64::try_from(parse_duration_ms(value)?)
+ .map_err(|_| "duration exceeds the supported range".to_string())?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.dynamic-batch-size.enabled",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_dynamic_batch_size_enabled = parse_bool(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.dynamic-batch-size.min",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_dynamic_batch_size_min = parse_i32_bytes(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.kv-backpressure.max-throttle",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_kv_backpressure_max_throttle_ms = parse_duration_ms(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "lookup.queue-size",
+ sensitive: false,
+ apply: |config, value| {
+ config.lookup_queue_size = parse_usize_count(value, 1)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "lookup.max-batch-size",
+ sensitive: false,
+ apply: |config, value| {
+ config.lookup_max_batch_size = parse_usize_count(value, 1)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "lookup.max-inflight-requests",
+ sensitive: false,
+ apply: |config, value| {
+ config.lookup_max_inflight_requests = parse_usize_count(value, 1)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "lookup.max-retries",
+ sensitive: false,
+ apply: |config, value| {
+ config.lookup_max_retries = parse_i32_count(value, 0)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "lookup.batch-timeout",
+ sensitive: false,
+ apply: |config, value| {
+ config.lookup_batch_timeout_ms = parse_duration_ms(value)?;
+ Ok(())
+ },
+ },
+];
+
+fn config_property_spec(key: &str) -> Option<&'static ConfigPropertySpec> {
+ CONFIG_PROPERTY_SPECS.iter().find(|spec| spec.key == key)
+}
+
+fn parse_bool(value: &str) -> Result<bool, String> {
+ value
+ .parse::<bool>()
+ .map_err(|_| "expected 'true' or 'false'".to_string())
+}
+
+fn parse_i32_count(value: &str, min: i32) -> Result<i32, String> {
+ let parsed = value
+ .parse::<i32>()
+ .map_err(|_| "expected a 32-bit integer".to_string())?;
+ if parsed < min {
+ return Err(format!("must be at least {min}"));
+ }
+ Ok(parsed)
+}
+
+fn parse_usize_count(value: &str, min: usize) -> Result<usize, String> {
+ let parsed = value
+ .parse::<usize>()
+ .map_err(|_| "expected a non-negative integer".to_string())?;
+ if parsed < min {
+ return Err(format!("must be at least {min}"));
+ }
+ Ok(parsed)
+}
+
+fn parse_i32_bytes(value: &str) -> Result<i32, String> {
+ i32::try_from(parse_byte_size(value)?)
+ .map_err(|_| "byte size exceeds the supported range".to_string())
+}
+
+fn parse_usize_bytes(value: &str) -> Result<usize, String> {
+ usize::try_from(parse_byte_size(value)?)
+ .map_err(|_| "byte size exceeds the supported range".to_string())
+}
+
+fn parse_byte_size(value: &str) -> Result<u64, String> {
+ let (number, unit) = split_number_and_unit(value)?;
+ let multiplier = match unit.to_ascii_lowercase().as_str() {
+ "" | "b" | "bytes" => 1,
+ "k" | "kb" | "kib" | "kibibyte" | "kibibytes" => 1_024,
+ "m" | "mb" | "mib" | "mebibyte" | "mebibytes" => 1_024 * 1_024,
+ "g" | "gb" | "gib" | "gibibyte" | "gibibytes" => 1_024 * 1_024 * 1_024,
+ "t" | "tb" | "tib" | "tebibyte" | "tebibytes" => 1_024_u64 * 1_024 * 1_024 * 1_024,
+ _ => return Err("unsupported byte-size unit".to_string()),
+ };
+ let bytes = number
+ .checked_mul(multiplier)
+ .ok_or_else(|| "byte size exceeds the supported range".to_string())?;
+ if bytes == 0 {
+ return Err("byte size must be greater than zero".to_string());
+ }
+ Ok(bytes)
+}
+
+fn parse_duration_ms(value: &str) -> Result<u64, String> {
Review Comment:
Could zero be validated per property rather than by the generic duration parser? `lookup.batch-timeout=0ms` currently passes validation; `LookupQueue::drain()` then exits before `recv()`, leaving requests queued while `LookupSender` can spin on empty batches. Please reject zero for connect/lookup timeouts and add a runtime regression test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rust] Add property overlays for rust client Config [fluss]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #4068:
URL: https://github.com/apache/fluss/pull/4068#discussion_r3844796418
##########
fluss-rust/crates/fluss/src/config.rs:
##########
@@ -212,6 +213,250 @@ pub struct Config {
pub lookup_max_retries: i32,
}
+type ConfigPropertyApplier = fn(&mut Config, &str) -> Result<(), String>;
+
+struct ConfigPropertySpec {
+ key: &'static str,
+ sensitive: bool,
+ apply: ConfigPropertyApplier,
+}
+
+const CONFIG_PROPERTY_SPECS: &[ConfigPropertySpec] = &[
Review Comment:
Only some of the Config fields have a property here. bootstrap.servers is the one I'd expect first: it's in every example we ship, and the gateway sets it too, so the gateway couldn't use this API without it.
All the scanner.* options are missing as well, yet validate() still validates the scanner settings.
Should we cover the rest?
##########
fluss-rust/crates/fluss/src/config.rs:
##########
@@ -212,6 +213,250 @@ pub struct Config {
pub lookup_max_retries: i32,
}
+type ConfigPropertyApplier = fn(&mut Config, &str) -> Result<(), String>;
+
+struct ConfigPropertySpec {
+ key: &'static str,
+ sensitive: bool,
+ apply: ConfigPropertyApplier,
+}
+
+const CONFIG_PROPERTY_SPECS: &[ConfigPropertySpec] = &[
+ ConfigPropertySpec {
+ key: "security.protocol",
+ sensitive: false,
+ apply: |config, value| {
+ config.security_protocol = value.to_string();
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "security.sasl.mechanism",
+ sensitive: false,
+ apply: |config, value| {
+ config.security_sasl_mechanism = value.to_string();
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "security.sasl.username",
+ sensitive: false,
+ apply: |config, value| {
+ config.security_sasl_username = value.to_string();
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "security.sasl.password",
+ sensitive: true,
+ apply: |config, value| {
+ config.security_sasl_password = value.to_string();
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "connect-timeout",
+ sensitive: false,
+ apply: |config, value| {
+ config.connect_timeout_ms = parse_duration_ms(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.batch-size",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_batch_size = parse_i32_bytes(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.request-max-size",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_request_max_size = parse_i32_bytes(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.buffer.memory-size",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_buffer_memory_size = parse_usize_bytes(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.buffer.wait-timeout",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_buffer_wait_timeout_ms = parse_duration_ms(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.batch-timeout",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_batch_timeout_ms = i64::try_from(parse_duration_ms(value)?)
+ .map_err(|_| "duration exceeds the supported range".to_string())?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.dynamic-batch-size.enabled",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_dynamic_batch_size_enabled = parse_bool(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.dynamic-batch-size.min",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_dynamic_batch_size_min = parse_i32_bytes(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.kv-backpressure.max-throttle",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_kv_backpressure_max_throttle_ms = parse_duration_ms(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "lookup.queue-size",
+ sensitive: false,
+ apply: |config, value| {
+ config.lookup_queue_size = parse_usize_count(value, 1)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "lookup.max-batch-size",
+ sensitive: false,
+ apply: |config, value| {
+ config.lookup_max_batch_size = parse_usize_count(value, 1)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "lookup.max-inflight-requests",
+ sensitive: false,
+ apply: |config, value| {
+ config.lookup_max_inflight_requests = parse_usize_count(value, 1)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "lookup.max-retries",
+ sensitive: false,
+ apply: |config, value| {
+ config.lookup_max_retries = parse_i32_count(value, 0)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "lookup.batch-timeout",
+ sensitive: false,
+ apply: |config, value| {
+ config.lookup_batch_timeout_ms = parse_duration_ms(value)?;
+ Ok(())
+ },
+ },
+];
+
+fn config_property_spec(key: &str) -> Option<&'static ConfigPropertySpec> {
+ CONFIG_PROPERTY_SPECS.iter().find(|spec| spec.key == key)
+}
+
+fn parse_bool(value: &str) -> Result<bool, String> {
+ value
+ .parse::<bool>()
+ .map_err(|_| "expected 'true' or 'false'".to_string())
+}
+
+fn parse_i32_count(value: &str, min: i32) -> Result<i32, String> {
+ let parsed = value
+ .parse::<i32>()
+ .map_err(|_| "expected a 32-bit integer".to_string())?;
+ if parsed < min {
+ return Err(format!("must be at least {min}"));
+ }
+ Ok(parsed)
+}
+
+fn parse_usize_count(value: &str, min: usize) -> Result<usize, String> {
+ let parsed = value
+ .parse::<usize>()
+ .map_err(|_| "expected a non-negative integer".to_string())?;
+ if parsed < min {
+ return Err(format!("must be at least {min}"));
+ }
+ Ok(parsed)
+}
+
+fn parse_i32_bytes(value: &str) -> Result<i32, String> {
+ i32::try_from(parse_byte_size(value)?)
+ .map_err(|_| "byte size exceeds the supported range".to_string())
+}
+
+fn parse_usize_bytes(value: &str) -> Result<usize, String> {
+ usize::try_from(parse_byte_size(value)?)
+ .map_err(|_| "byte size exceeds the supported range".to_string())
+}
+
+fn parse_byte_size(value: &str) -> Result<u64, String> {
Review Comment:
This is the same as ByteSize::parse in fluss-gateway/src/config.rs: same units, same zero check, and split_number_and_unit even has the same name.
The gateway already depends on this crate. Could the parser live here once and the gateway call it?
Separately, these units don't match Java. MemorySize rejects 4mib, 1kib and 1gib, rejects a bare 512 because a unit is required, and accepts 0b.
Being looser than Java is fine if it's deliberate, but two copies that can drift apart isn't.
##########
fluss-rust/crates/fluss/src/config.rs:
##########
@@ -212,6 +213,250 @@ pub struct Config {
pub lookup_max_retries: i32,
}
+type ConfigPropertyApplier = fn(&mut Config, &str) -> Result<(), String>;
+
+struct ConfigPropertySpec {
+ key: &'static str,
+ sensitive: bool,
+ apply: ConfigPropertyApplier,
+}
+
+const CONFIG_PROPERTY_SPECS: &[ConfigPropertySpec] = &[
+ ConfigPropertySpec {
+ key: "security.protocol",
+ sensitive: false,
+ apply: |config, value| {
+ config.security_protocol = value.to_string();
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "security.sasl.mechanism",
+ sensitive: false,
+ apply: |config, value| {
+ config.security_sasl_mechanism = value.to_string();
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "security.sasl.username",
+ sensitive: false,
+ apply: |config, value| {
+ config.security_sasl_username = value.to_string();
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "security.sasl.password",
+ sensitive: true,
Review Comment:
Is this flag meant to be read somewhere? is_sensitive_property has no callers, and Debug already prints [REDACTED] for the password on line 527. Could we drop it for now and use the gateway's Secret type if we need redaction later?
And is test_property_errors_do_not_expose_sensitive_values testing what it says?
It passes with a non-sensitive key too, since errors never include the value for any key.
##########
fluss-rust/crates/fluss/src/config.rs:
##########
@@ -333,6 +578,36 @@ impl Default for Config {
}
impl Config {
+ /// Applies canonical string properties to a cloned config. Cross-field
+ /// validation is deferred to [`Config::validate`].
+ pub fn try_with_properties(
Review Comment:
Nothing calls this yet, so could we convert one caller here to show it works?
native_client_config() at fluss-gateway/src/config.rs builds a Config by hand and looks like the easy one, though it needs bootstrap.servers first. Or would you rather follow up separately?
Minor: would impl IntoIterator be easier to call than &BTreeMap<String, String>?
##########
fluss-rust/crates/fluss/src/config.rs:
##########
@@ -212,6 +213,250 @@ pub struct Config {
pub lookup_max_retries: i32,
}
+type ConfigPropertyApplier = fn(&mut Config, &str) -> Result<(), String>;
+
+struct ConfigPropertySpec {
+ key: &'static str,
+ sensitive: bool,
+ apply: ConfigPropertyApplier,
+}
+
+const CONFIG_PROPERTY_SPECS: &[ConfigPropertySpec] = &[
+ ConfigPropertySpec {
+ key: "security.protocol",
+ sensitive: false,
+ apply: |config, value| {
+ config.security_protocol = value.to_string();
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "security.sasl.mechanism",
+ sensitive: false,
+ apply: |config, value| {
+ config.security_sasl_mechanism = value.to_string();
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "security.sasl.username",
+ sensitive: false,
+ apply: |config, value| {
+ config.security_sasl_username = value.to_string();
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "security.sasl.password",
+ sensitive: true,
+ apply: |config, value| {
+ config.security_sasl_password = value.to_string();
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "connect-timeout",
+ sensitive: false,
+ apply: |config, value| {
+ config.connect_timeout_ms = parse_duration_ms(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.batch-size",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_batch_size = parse_i32_bytes(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.request-max-size",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_request_max_size = parse_i32_bytes(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.buffer.memory-size",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_buffer_memory_size = parse_usize_bytes(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.buffer.wait-timeout",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_buffer_wait_timeout_ms = parse_duration_ms(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.batch-timeout",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_batch_timeout_ms = i64::try_from(parse_duration_ms(value)?)
+ .map_err(|_| "duration exceeds the supported range".to_string())?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.dynamic-batch-size.enabled",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_dynamic_batch_size_enabled = parse_bool(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.dynamic-batch-size.min",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_dynamic_batch_size_min = parse_i32_bytes(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "writer.kv-backpressure.max-throttle",
+ sensitive: false,
+ apply: |config, value| {
+ config.writer_kv_backpressure_max_throttle_ms = parse_duration_ms(value)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "lookup.queue-size",
+ sensitive: false,
+ apply: |config, value| {
+ config.lookup_queue_size = parse_usize_count(value, 1)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "lookup.max-batch-size",
+ sensitive: false,
+ apply: |config, value| {
+ config.lookup_max_batch_size = parse_usize_count(value, 1)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "lookup.max-inflight-requests",
+ sensitive: false,
+ apply: |config, value| {
+ config.lookup_max_inflight_requests = parse_usize_count(value, 1)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "lookup.max-retries",
+ sensitive: false,
+ apply: |config, value| {
+ config.lookup_max_retries = parse_i32_count(value, 0)?;
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "lookup.batch-timeout",
+ sensitive: false,
+ apply: |config, value| {
+ config.lookup_batch_timeout_ms = parse_duration_ms(value)?;
+ Ok(())
+ },
+ },
+];
+
+fn config_property_spec(key: &str) -> Option<&'static ConfigPropertySpec> {
+ CONFIG_PROPERTY_SPECS.iter().find(|spec| spec.key == key)
+}
+
+fn parse_bool(value: &str) -> Result<bool, String> {
+ value
+ .parse::<bool>()
+ .map_err(|_| "expected 'true' or 'false'".to_string())
+}
+
+fn parse_i32_count(value: &str, min: i32) -> Result<i32, String> {
+ let parsed = value
+ .parse::<i32>()
+ .map_err(|_| "expected a 32-bit integer".to_string())?;
+ if parsed < min {
+ return Err(format!("must be at least {min}"));
+ }
+ Ok(parsed)
+}
+
+fn parse_usize_count(value: &str, min: usize) -> Result<usize, String> {
+ let parsed = value
+ .parse::<usize>()
+ .map_err(|_| "expected a non-negative integer".to_string())?;
+ if parsed < min {
+ return Err(format!("must be at least {min}"));
+ }
+ Ok(parsed)
+}
+
+fn parse_i32_bytes(value: &str) -> Result<i32, String> {
+ i32::try_from(parse_byte_size(value)?)
+ .map_err(|_| "byte size exceeds the supported range".to_string())
+}
+
+fn parse_usize_bytes(value: &str) -> Result<usize, String> {
+ usize::try_from(parse_byte_size(value)?)
+ .map_err(|_| "byte size exceeds the supported range".to_string())
+}
+
+fn parse_byte_size(value: &str) -> Result<u64, String> {
+ let (number, unit) = split_number_and_unit(value)?;
+ let multiplier = match unit.to_ascii_lowercase().as_str() {
+ "" | "b" | "bytes" => 1,
+ "k" | "kb" | "kib" | "kibibyte" | "kibibytes" => 1_024,
+ "m" | "mb" | "mib" | "mebibyte" | "mebibytes" => 1_024 * 1_024,
+ "g" | "gb" | "gib" | "gibibyte" | "gibibytes" => 1_024 * 1_024 * 1_024,
+ "t" | "tb" | "tib" | "tebibyte" | "tebibytes" => 1_024_u64 * 1_024 * 1_024 * 1_024,
+ _ => return Err("unsupported byte-size unit".to_string()),
+ };
+ let bytes = number
+ .checked_mul(multiplier)
+ .ok_or_else(|| "byte size exceeds the supported range".to_string())?;
+ if bytes == 0 {
+ return Err("byte size must be greater than zero".to_string());
+ }
+ Ok(bytes)
+}
+
+fn parse_duration_ms(value: &str) -> Result<u64, String> {
Review Comment:
Can we use the same unit list as Java's TimeUtils? This takes only ms/s/m/h/d, so values from our own docs don't parse - 3min, 1min, 1day are all defaults in website/docs/engine-flink/options.md, and TimeUtils accepts those plus sec, seconds, millis, ns and hour.
And should 0ms be accepted? It gives connect-timeout a zero timeout with nothing catching it. The gateway's duration parser rejects zero, should this one as well?
##########
fluss-rust/crates/fluss/src/config.rs:
##########
@@ -458,12 +733,214 @@ impl Config {
}
Ok(())
}
+
+ pub fn validate_lookup(&self) -> Result<(), String> {
Review Comment:
nit: Should these two share one message? Line 331 already rejects 0 with "must be at least 1", and this rejects it again with "lookup_queue_size must be > 0", so the same mistake reads differently depending on whether it came from a property or a CLI flag.
##########
fluss-rust/crates/fluss/src/config.rs:
##########
@@ -212,6 +213,250 @@ pub struct Config {
pub lookup_max_retries: i32,
}
+type ConfigPropertyApplier = fn(&mut Config, &str) -> Result<(), String>;
+
+struct ConfigPropertySpec {
+ key: &'static str,
+ sensitive: bool,
+ apply: ConfigPropertyApplier,
+}
+
+const CONFIG_PROPERTY_SPECS: &[ConfigPropertySpec] = &[
+ ConfigPropertySpec {
+ key: "security.protocol",
+ sensitive: false,
+ apply: |config, value| {
+ config.security_protocol = value.to_string();
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "security.sasl.mechanism",
+ sensitive: false,
+ apply: |config, value| {
+ config.security_sasl_mechanism = value.to_string();
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "security.sasl.username",
+ sensitive: false,
+ apply: |config, value| {
+ config.security_sasl_username = value.to_string();
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "security.sasl.password",
+ sensitive: true,
+ apply: |config, value| {
+ config.security_sasl_password = value.to_string();
+ Ok(())
+ },
+ },
+ ConfigPropertySpec {
+ key: "connect-timeout",
Review Comment:
Which value format should win for these keys? The Python binding already has them with different ones as there connect-timeout is plain milliseconds, so {"connect-timeout": "5000"} works today and "5s" errors, and api-reference.md documents it as milliseconds.
Under this parser it's reversed. Byte keys clash the same way: should writer.batch-size take "2mb", or only a bare integer as it does in Python today?
The binding will have to follow either way, so can we pin the format down here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org