This is an automated email from the ASF dual-hosted git repository.
leekeiabstraction pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss-rust.git
The following commit(s) were added to refs/heads/main by this push:
new db824d75 chore: Add validation for numeric config fields (#449)
db824d75 is described below
commit db824d751488ec618d77bc3b4866725d403b6867
Author: Prajwal banakar <[email protected]>
AuthorDate: Wed May 6 13:27:28 2026 +0530
chore: Add validation for numeric config fields (#449)
* Add validation for numeric config fields
* improved
* Added issue comments
* fix: resolve clippy empty-line-after-doc-comments warning
---
crates/fluss/src/client/connection.rs | 6 +-
crates/fluss/src/client/table/remote_log.rs | 2 +-
crates/fluss/src/client/write/writer_client.rs | 4 -
crates/fluss/src/config.rs | 213 +++++++++++++++++++------
4 files changed, 172 insertions(+), 53 deletions(-)
diff --git a/crates/fluss/src/client/connection.rs
b/crates/fluss/src/client/connection.rs
index a3ffd755..c31104c4 100644
--- a/crates/fluss/src/client/connection.rs
+++ b/crates/fluss/src/client/connection.rs
@@ -41,10 +41,14 @@ impl FlussConnection {
pub async fn new(arg: Config) -> Result<Self> {
arg.validate_security()
.map_err(|msg| Error::IllegalArgument { message: msg })?;
- arg.validate_scanner_fetch()
+ arg.validate_scanner()
+ .map_err(|msg| Error::IllegalArgument { message: msg })?;
+ arg.validate_writer()
.map_err(|msg| Error::IllegalArgument { message: msg })?;
let timeout = Duration::from_millis(arg.connect_timeout_ms);
+ // connect_timeout_ms: no lower-bound validation to match Java
behavior.
+ // Java allows 0 — tracked in
https://github.com/apache/fluss/issues/3068
let connections = if arg.is_sasl_enabled() {
Arc::new(
RpcClient::new()
diff --git a/crates/fluss/src/client/table/remote_log.rs
b/crates/fluss/src/client/table/remote_log.rs
index 6bc95512..4d96ce96 100644
--- a/crates/fluss/src/client/table/remote_log.rs
+++ b/crates/fluss/src/client/table/remote_log.rs
@@ -778,7 +778,7 @@ impl RemoteLogDownloader {
let fetcher = Arc::new(ProductionFetcher {
credentials_rx,
local_log_dir: Arc::new(local_log_dir),
- remote_log_read_concurrency: remote_log_read_concurrency.max(1),
+ remote_log_read_concurrency,
});
Self::new_with_fetcher(fetcher, max_prefetch_segments,
max_concurrent_downloads)
diff --git a/crates/fluss/src/client/write/writer_client.rs
b/crates/fluss/src/client/write/writer_client.rs
index aee6bcd9..ffdf96b1 100644
--- a/crates/fluss/src/client/write/writer_client.rs
+++ b/crates/fluss/src/client/write/writer_client.rs
@@ -54,10 +54,6 @@ impl WriterClient {
pub fn new(config: Config, metadata: Arc<Metadata>) -> Result<Self> {
let ack = Self::get_ack(&config)?;
- config
- .validate_idempotence()
- .map_err(|message| Error::IllegalArgument { message })?;
-
let idempotence_manager = Arc::new(IdempotenceManager::new(
config.writer_enable_idempotence,
config.writer_max_inflight_requests_per_bucket,
diff --git a/crates/fluss/src/config.rs b/crates/fluss/src/config.rs
index 2900e2f4..09a17f83 100644
--- a/crates/fluss/src/config.rs
+++ b/crates/fluss/src/config.rs
@@ -303,39 +303,6 @@ impl Config {
pub fn is_sasl_enabled(&self) -> bool {
self.security_protocol.eq_ignore_ascii_case("sasl")
}
-
- /// Validates idempotence configuration. Returns `Ok(())` when the config
is
- /// consistent, or an error message when idempotence is enabled but other
- /// settings are incompatible.
- pub fn validate_idempotence(&self) -> Result<(), String> {
- if !self.writer_enable_idempotence {
- return Ok(());
- }
- let acks_is_all = self.writer_acks.eq_ignore_ascii_case("all") ||
self.writer_acks == "-1";
- if !acks_is_all {
- return Err(format!(
- "Idempotent writes require acks='all' (-1), but got acks='{}'",
- self.writer_acks
- ));
- }
- if self.writer_retries <= 0 {
- return Err(format!(
- "Idempotent writes require retries > 0, but got retries={}",
- self.writer_retries
- ));
- }
- if self.writer_max_inflight_requests_per_bucket
- > MAX_IN_FLIGHT_REQUESTS_PER_BUCKET_FOR_IDEMPOTENCE
- {
- return Err(format!(
- "Idempotent writes require max-inflight-requests-per-bucket <=
{}, but got {}",
- MAX_IN_FLIGHT_REQUESTS_PER_BUCKET_FOR_IDEMPOTENCE,
- self.writer_max_inflight_requests_per_bucket
- ));
- }
- Ok(())
- }
-
/// Validates security configuration. Returns `Ok(())` when the config is
/// consistent, or an error message when SASL is enabled but the config is
/// incomplete or uses an unsupported mechanism.
@@ -361,7 +328,18 @@ impl Config {
}
Ok(())
}
- pub fn validate_scanner_fetch(&self) -> Result<(), String> {
+ pub fn validate_scanner(&self) -> Result<(), String> {
+ if self.scanner_remote_log_prefetch_num == 0 {
+ return Err("scanner_remote_log_prefetch_num must be >
0".to_string());
+ }
+ if self.scanner_remote_log_read_concurrency == 0 {
+ return Err("scanner_remote_log_read_concurrency must be >
0".to_string());
+ }
+ if self.remote_file_download_thread_num == 0 {
+ return Err("remote_file_download_thread_num must be >
0".to_string());
+ }
+ // scanner_log_max_poll_records: validation intentionally omitted to
match Java behavior.
+ // Java allows 0 — tracked in
https://github.com/apache/fluss/issues/3068
if self.scanner_log_fetch_min_bytes <= 0 {
return Err("scanner_log_fetch_min_bytes must be > 0".to_string());
}
@@ -387,6 +365,57 @@ impl Config {
}
Ok(())
}
+
+ pub fn validate_writer(&self) -> Result<(), String> {
+ if self.writer_request_max_size <= 0 {
+ return Err("writer_request_max_size must be > 0".to_string());
+ }
+ if self.writer_batch_size <= 0 {
+ return Err("writer_batch_size must be > 0".to_string());
+ }
+ if self.writer_batch_timeout_ms < 0 {
+ return Err("writer_batch_timeout_ms must be >= 0".to_string());
+ }
+ if self.writer_max_inflight_requests_per_bucket == 0 {
+ return Err("writer_max_inflight_requests_per_bucket must be >
0".to_string());
+ }
+ if self.writer_buffer_memory_size == 0 {
+ return Err("writer_buffer_memory_size must be > 0".to_string());
+ }
+ if self.writer_batch_size > self.writer_request_max_size {
+ return Err("writer_batch_size must be <=
writer_request_max_size".to_string());
+ }
+ if self.writer_batch_size as usize > self.writer_buffer_memory_size {
+ return Err("writer_batch_size must be <=
writer_buffer_memory_size".to_string());
+ }
+ // idempotence checks
+ if !self.writer_enable_idempotence {
+ return Ok(());
+ }
+ let acks_is_all = self.writer_acks.eq_ignore_ascii_case("all") ||
self.writer_acks == "-1";
+ if !acks_is_all {
+ return Err(format!(
+ "Idempotent writes require acks='all' (-1), but got acks='{}'",
+ self.writer_acks
+ ));
+ }
+ if self.writer_retries <= 0 {
+ return Err(format!(
+ "Idempotent writes require retries > 0, but got retries={}",
+ self.writer_retries
+ ));
+ }
+ if self.writer_max_inflight_requests_per_bucket
+ > MAX_IN_FLIGHT_REQUESTS_PER_BUCKET_FOR_IDEMPOTENCE
+ {
+ return Err(format!(
+ "Idempotent writes require max-inflight-requests-per-bucket <=
{}, but got {}",
+ MAX_IN_FLIGHT_REQUESTS_PER_BUCKET_FOR_IDEMPOTENCE,
+ self.writer_max_inflight_requests_per_bucket
+ ));
+ }
+ Ok(())
+ }
}
#[cfg(test)]
@@ -456,13 +485,38 @@ mod tests {
};
assert!(config.validate_security().is_err());
}
+
#[test]
- fn test_scanner_fetch_defaults_valid() {
+ fn test_scanner_defaults_valid() {
let config = Config::default();
- assert!(config.validate_scanner_fetch().is_ok());
- assert_eq!(config.scanner_log_fetch_max_bytes, 16 * 1024 * 1024);
- assert_eq!(config.scanner_log_fetch_min_bytes, 1);
- assert_eq!(config.scanner_log_fetch_wait_max_time_ms, 500);
+ assert!(config.validate_scanner().is_ok());
+ }
+
+ #[test]
+ fn test_scanner_remote_log_prefetch_num_zero() {
+ let config = Config {
+ scanner_remote_log_prefetch_num: 0,
+ ..Config::default()
+ };
+ assert!(config.validate_scanner().is_err());
+ }
+
+ #[test]
+ fn test_scanner_remote_log_read_concurrency_zero() {
+ let config = Config {
+ scanner_remote_log_read_concurrency: 0,
+ ..Config::default()
+ };
+ assert!(config.validate_scanner().is_err());
+ }
+
+ #[test]
+ fn test_remote_file_download_thread_num_zero() {
+ let config = Config {
+ remote_file_download_thread_num: 0,
+ ..Config::default()
+ };
+ assert!(config.validate_scanner().is_err());
}
#[test]
@@ -472,7 +526,7 @@ mod tests {
scanner_log_fetch_max_bytes: 1,
..Config::default()
};
- assert!(config.validate_scanner_fetch().is_err());
+ assert!(config.validate_scanner().is_err());
}
#[test]
@@ -481,13 +535,78 @@ mod tests {
scanner_log_fetch_wait_max_time_ms: -1,
..Config::default()
};
- assert!(config.validate_scanner_fetch().is_err());
+ assert!(config.validate_scanner().is_err());
}
#[test]
- fn test_idempotence_default_is_valid() {
+ fn test_writer_defaults_valid() {
let config = Config::default();
- assert!(config.validate_idempotence().is_ok());
+ assert!(config.validate_writer().is_ok());
+ }
+
+ #[test]
+ fn test_writer_request_max_size_zero() {
+ let config = Config {
+ writer_request_max_size: 0,
+ ..Config::default()
+ };
+ assert!(config.validate_writer().is_err());
+ }
+
+ #[test]
+ fn test_writer_batch_size_zero() {
+ let config = Config {
+ writer_batch_size: 0,
+ ..Config::default()
+ };
+ assert!(config.validate_writer().is_err());
+ }
+
+ #[test]
+ fn test_writer_batch_timeout_negative() {
+ let config = Config {
+ writer_batch_timeout_ms: -1,
+ ..Config::default()
+ };
+ assert!(config.validate_writer().is_err());
+ }
+
+ #[test]
+ fn test_writer_max_inflight_requests_per_bucket_zero() {
+ let config = Config {
+ writer_max_inflight_requests_per_bucket: 0,
+ ..Config::default()
+ };
+ assert!(config.validate_writer().is_err());
+ }
+
+ #[test]
+ fn test_writer_buffer_memory_size_zero() {
+ let config = Config {
+ writer_buffer_memory_size: 0,
+ ..Config::default()
+ };
+ assert!(config.validate_writer().is_err());
+ }
+
+ #[test]
+ fn test_writer_batch_size_exceeds_request_max_size() {
+ let config = Config {
+ writer_batch_size: 20 * 1024 * 1024,
+ writer_request_max_size: 10 * 1024 * 1024,
+ ..Config::default()
+ };
+ assert!(config.validate_writer().is_err());
+ }
+
+ #[test]
+ fn test_writer_batch_size_exceeds_buffer_memory_size() {
+ let config = Config {
+ writer_batch_size: 128 * 1024 * 1024,
+ writer_buffer_memory_size: 64 * 1024 * 1024,
+ ..Config::default()
+ };
+ assert!(config.validate_writer().is_err());
}
#[test]
@@ -499,7 +618,7 @@ mod tests {
writer_max_inflight_requests_per_bucket: 100,
..Config::default()
};
- assert!(config.validate_idempotence().is_ok());
+ assert!(config.validate_writer().is_ok());
}
#[test]
@@ -509,7 +628,7 @@ mod tests {
writer_acks: "1".to_string(),
..Config::default()
};
- assert!(config.validate_idempotence().is_err());
+ assert!(config.validate_writer().is_err());
}
#[test]
@@ -519,7 +638,7 @@ mod tests {
writer_retries: 0,
..Config::default()
};
- assert!(config.validate_idempotence().is_err());
+ assert!(config.validate_writer().is_err());
}
#[test]
@@ -529,6 +648,6 @@ mod tests {
writer_max_inflight_requests_per_bucket: 10,
..Config::default()
};
- assert!(config.validate_idempotence().is_err());
+ assert!(config.validate_writer().is_err());
}
}
