(fluss) branch main updated: [client] Add validation for client.scanner.log.max-poll-records and client.connect-timeout (#3069)

[aborisov]

This is an automated email from the ASF dual-hosted git repository.

fresh-borzoni pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git


The following commit(s) were added to refs/heads/main by this push:
     new 486fe0a8a [client] Add validation for 
client.scanner.log.max-poll-records and client.connect-timeout (#3069)
486fe0a8a is described below

commit 486fe0a8ae8c0677c468b94dfdfec39535453f36
Author: Prajwal Banakar <prajwal.banak...@gmail.com>
AuthorDate: Fri Jun 26 19:09:35 2026 +0530

    [client] Add validation for client.scanner.log.max-poll-records and 
client.connect-timeout (#3069)
    
    * [client] Add validation for client.scanner.log.max-poll-records and 
client.connect-timeout
    
    * adress review
---
 .../org/apache/fluss/client/FlussConnection.java   |  2 ++
 .../org/apache/fluss/config/FlussConfigUtils.java  | 17 +++++++++++++++
 .../apache/fluss/config/FlussConfigUtilsTest.java  | 24 ++++++++++++++++++++++
 3 files changed, 43 insertions(+)

diff --git 
a/fluss-client/src/main/java/org/apache/fluss/client/FlussConnection.java 
b/fluss-client/src/main/java/org/apache/fluss/client/FlussConnection.java
index e5eaa41c3..0d800b3c9 100644
--- a/fluss-client/src/main/java/org/apache/fluss/client/FlussConnection.java
+++ b/fluss-client/src/main/java/org/apache/fluss/client/FlussConnection.java
@@ -32,6 +32,7 @@ import org.apache.fluss.client.write.WriterClient;
 import org.apache.fluss.cluster.ServerNode;
 import org.apache.fluss.config.ConfigOptions;
 import org.apache.fluss.config.Configuration;
+import org.apache.fluss.config.FlussConfigUtils;
 import org.apache.fluss.exception.FlussRuntimeException;
 import org.apache.fluss.fs.FileSystem;
 import org.apache.fluss.metadata.TablePath;
@@ -70,6 +71,7 @@ public final class FlussConnection implements Connection {
 
     FlussConnection(Configuration conf, MetricRegistry metricRegistry) {
         this.conf = conf;
+        FlussConfigUtils.validateClientConfigs(conf);
         // init Filesystem with configuration from FlussConnection,
         // only pass options with 'client.fs.' prefix
         FileSystem.initialize(
diff --git 
a/fluss-common/src/main/java/org/apache/fluss/config/FlussConfigUtils.java 
b/fluss-common/src/main/java/org/apache/fluss/config/FlussConfigUtils.java
index 07334df7b..4f978a0b3 100644
--- a/fluss-common/src/main/java/org/apache/fluss/config/FlussConfigUtils.java
+++ b/fluss-common/src/main/java/org/apache/fluss/config/FlussConfigUtils.java
@@ -23,6 +23,7 @@ import 
org.apache.fluss.exception.IllegalConfigurationException;
 import org.apache.fluss.fs.FsPath;
 
 import java.lang.reflect.Field;
+import java.time.Duration;
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
@@ -241,4 +242,20 @@ public class FlussConfigUtils {
                             option.key(), minValue));
         }
     }
+
+    public static void validateClientConfigs(Configuration conf) {
+        validMinValue(conf, ConfigOptions.CLIENT_SCANNER_LOG_MAX_POLL_RECORDS, 
1);
+        validMinDuration(conf, ConfigOptions.CLIENT_CONNECT_TIMEOUT, 1);
+    }
+
+    private static void validMinDuration(
+            Configuration conf, ConfigOption<Duration> option, long minMillis) 
{
+        long millis = conf.get(option).toMillis();
+        if (millis < minMillis) {
+            throw new IllegalConfigurationException(
+                    String.format(
+                            "Invalid configuration for %s, it must be greater 
than or equal %d ms.",
+                            option.key(), minMillis));
+        }
+    }
 }
diff --git 
a/fluss-common/src/test/java/org/apache/fluss/config/FlussConfigUtilsTest.java 
b/fluss-common/src/test/java/org/apache/fluss/config/FlussConfigUtilsTest.java
index 2783eea7a..f5e4b20ca 100644
--- 
a/fluss-common/src/test/java/org/apache/fluss/config/FlussConfigUtilsTest.java
+++ 
b/fluss-common/src/test/java/org/apache/fluss/config/FlussConfigUtilsTest.java
@@ -21,6 +21,7 @@ import 
org.apache.fluss.exception.IllegalConfigurationException;
 
 import org.junit.jupiter.api.Test;
 
+import java.time.Duration;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.Map;
@@ -189,4 +190,27 @@ class FlussConfigUtilsTest {
                 .hasMessageContaining(ConfigOptions.TABLET_SERVER_ID.key())
                 .hasMessageContaining("it must be greater than or equal 0");
     }
+
+    @Test
+    void testValidateClientConfigs() {
+        // valid defaults should pass
+        Configuration validConf = new Configuration();
+        FlussConfigUtils.validateClientConfigs(validConf);
+
+        // max-poll-records = 0 should fail
+        Configuration zeroPollConf = new Configuration();
+        zeroPollConf.set(ConfigOptions.CLIENT_SCANNER_LOG_MAX_POLL_RECORDS, 0);
+        assertThatThrownBy(() -> 
FlussConfigUtils.validateClientConfigs(zeroPollConf))
+                .isInstanceOf(IllegalConfigurationException.class)
+                
.hasMessageContaining(ConfigOptions.CLIENT_SCANNER_LOG_MAX_POLL_RECORDS.key())
+                .hasMessageContaining("must be greater than or equal 1");
+
+        // connect-timeout = 0 should fail
+        Configuration zeroTimeoutConf = new Configuration();
+        zeroTimeoutConf.set(ConfigOptions.CLIENT_CONNECT_TIMEOUT, 
Duration.ZERO);
+        assertThatThrownBy(() -> 
FlussConfigUtils.validateClientConfigs(zeroTimeoutConf))
+                .isInstanceOf(IllegalConfigurationException.class)
+                
.hasMessageContaining(ConfigOptions.CLIENT_CONNECT_TIMEOUT.key())
+                .hasMessageContaining("must be greater than or equal 1 ms");
+    }
 }