[client] Add validation for client.scanner.log.max-poll-records and client.connect-timeout

fluss-client/src/main/java/org/apache/fluss/client/FlussConnection.java

@@ -32,6 +32,7 @@
 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(

fluss-common/src/main/java/org/apache/fluss/config/FlussConfigUtils.java

@@ -23,6 +23,7 @@
 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;
@@ -225,4 +226,20 @@ private static void validMinValue(ConfigOption<Integer> option, int value, int m
                             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));
+        }
+    }
 }

fluss-common/src/test/java/org/apache/fluss/config/FlussConfigUtilsTest.java

@@ -21,6 +21,7 @@
 
 import org.junit.jupiter.api.Test;
 
+import java.time.Duration;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.Map;
@@ -175,4 +176,27 @@ void testValidateTabletConfigs() {
                 .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");
+    }
 }