[Improve][Connector-V2] Migrate MQTT connector validation to declarative OptionRule

seatunnel-connectors-v2/connector-mqtt/src/main/java/org/apache/seatunnel/connectors/seatunnel/mqtt/config/MqttFormatValidator.java

@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.mqtt.config;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.configuration.util.ConditionExtension;
+
+/**
+ * Validates that the MQTT message {@code format} option is one of the supported values ({@code
+ * json} or {@code text}, case-insensitive). Shared by the MQTT source and sink option rules.
+ */
+public class MqttFormatValidator implements ConditionExtension<String> {
+
+    @Override
+    public String description() {
+        return "must be one of [json, text] (case-insensitive)";
+    }
+
+    @Override
+    public boolean evaluate(ReadonlyConfig config, String value) {
+        return "json".equalsIgnoreCase(value) || "text".equalsIgnoreCase(value);
+    }
+}

seatunnel-connectors-v2/connector-mqtt/src/main/java/org/apache/seatunnel/connectors/seatunnel/mqtt/sink/MqttSinkFactory.java

@@ -17,11 +17,13 @@
 
 package org.apache.seatunnel.connectors.seatunnel.mqtt.sink;
 
+import org.apache.seatunnel.api.configuration.util.Conditions;
 import org.apache.seatunnel.api.configuration.util.OptionRule;
 import org.apache.seatunnel.api.table.connector.TableSink;
 import org.apache.seatunnel.api.table.factory.Factory;
 import org.apache.seatunnel.api.table.factory.TableSinkFactory;
 import org.apache.seatunnel.api.table.factory.TableSinkFactoryContext;
+import org.apache.seatunnel.connectors.seatunnel.mqtt.config.MqttFormatValidator;
 
 import com.google.auto.service.AutoService;
 
@@ -40,13 +42,20 @@ public OptionRule optionRule() {
                 .optional(
                         MqttSinkOptions.USERNAME,
                         MqttSinkOptions.PASSWORD,
-                        MqttSinkOptions.QOS,
-                        MqttSinkOptions.FORMAT,
                         MqttSinkOptions.FIELD_DELIMITER,
-                        MqttSinkOptions.BATCH_SIZE,
                         MqttSinkOptions.RETRY_TIMEOUT,
                         MqttSinkOptions.CONNECTION_TIMEOUT,
                         MqttSinkOptions.CLEAN_SESSION)
+                .optional(
+                        MqttSinkOptions.QOS,
+                        Conditions.greaterOrEqual(MqttSinkOptions.QOS, 0)
+                                .and(Conditions.lessOrEqual(MqttSinkOptions.QOS, 1)))
+                .optional(
+                        MqttSinkOptions.FORMAT,
+                        Conditions.extension(MqttSinkOptions.FORMAT, new MqttFormatValidator()))
+                .optional(
+                        MqttSinkOptions.BATCH_SIZE,
+                        Conditions.greaterOrEqual(MqttSinkOptions.BATCH_SIZE, 1))
                 .build();
     }
 

seatunnel-connectors-v2/connector-mqtt/src/main/java/org/apache/seatunnel/connectors/seatunnel/mqtt/sink/MqttSinkWriter.java

@@ -63,17 +63,12 @@ public class MqttSinkWriter implements SinkWriter<SeaTunnelRow, Void, Void>, Mqt
 
     public MqttSinkWriter(
             SinkWriter.Context context, SeaTunnelRowType rowType, ReadonlyConfig pluginConfig) {
+        // Option value validation (qos range, batch_size, format) is enforced declaratively by
+        // MqttSinkFactory#optionRule() at job submission time.
         this.topic = pluginConfig.get(MqttSinkOptions.TOPIC);
         this.qos = pluginConfig.get(MqttSinkOptions.QOS);
-        if (this.qos < 0 || this.qos > 1) {
-            throw new IllegalArgumentException(
-                    "MQTT QoS must be 0 (at-most-once) or 1 (at-least-once), got: " + this.qos);
-        }
         this.retryTimeoutMs = pluginConfig.get(MqttSinkOptions.RETRY_TIMEOUT);
         this.batchSize = pluginConfig.get(MqttSinkOptions.BATCH_SIZE);
-        if (this.batchSize < 1) {
-            throw new IllegalArgumentException("batch_size must be >= 1, got: " + this.batchSize);
-        }
         this.messageBuffer = new ArrayList<>(this.batchSize);
         this.serializationSchema = createSerializationSchema(rowType, pluginConfig);
 

seatunnel-connectors-v2/connector-mqtt/src/main/java/org/apache/seatunnel/connectors/seatunnel/mqtt/source/MqttSourceConfig.java

@@ -53,34 +53,14 @@ public MqttSourceConfig(ReadonlyConfig config) {
         this.reconnectTimeout = config.get(MqttSourceOptions.RECONNECT_TIMEOUT);
         this.maxQueueSize = config.get(MqttSourceOptions.MAX_QUEUE_SIZE);
 
+        // Option value validation (qos range, format, reconnect_timeout, max_queue_size,
+        // client_id requirement for persistent sessions) is enforced declaratively by
+        // MqttSourceFactory#optionRule() at job submission time.
         String configuredClientId = config.get(MqttSourceOptions.CLIENT_ID);
-        if (!cleanSession && isBlank(configuredClientId)) {
-            throw new IllegalArgumentException(
-                    "client_id is required when clean_session=false for MQTT source");
-        }
         this.clientId =
                 isBlank(configuredClientId)
                         ? CLIENT_ID_PREFIX + UUID.randomUUID().toString()
                         : configuredClientId;
-
-        validate();
-    }
-
-    private void validate() {
-        if (qos < 0 || qos > 1) {
-            throw new IllegalArgumentException("MQTT source qos must be 0 or 1, got: " + qos);
-        }
-        if (!"json".equalsIgnoreCase(format) && !"text".equalsIgnoreCase(format)) {
-            throw new IllegalArgumentException("Unsupported MQTT source format: " + format);
-        }
-        if (reconnectTimeout <= 0) {
-            throw new IllegalArgumentException(
-                    "reconnect_timeout must be greater than 0, got: " + reconnectTimeout);
-        }
-        if (maxQueueSize <= 0) {
-            throw new IllegalArgumentException(
-                    "max_queue_size must be greater than 0, got: " + maxQueueSize);
-        }
     }
 
     private static boolean isBlank(String value) {

seatunnel-connectors-v2/connector-mqtt/src/main/java/org/apache/seatunnel/connectors/seatunnel/mqtt/source/MqttSourceFactory.java

@@ -17,6 +17,7 @@
 
 package org.apache.seatunnel.connectors.seatunnel.mqtt.source;
 
+import org.apache.seatunnel.api.configuration.util.Conditions;
 import org.apache.seatunnel.api.configuration.util.OptionRule;
 import org.apache.seatunnel.api.options.ConnectorCommonOptions;
 import org.apache.seatunnel.api.source.SeaTunnelSource;
@@ -25,6 +26,7 @@
 import org.apache.seatunnel.api.table.factory.Factory;
 import org.apache.seatunnel.api.table.factory.TableSourceFactory;
 import org.apache.seatunnel.api.table.factory.TableSourceFactoryContext;
+import org.apache.seatunnel.connectors.seatunnel.mqtt.config.MqttFormatValidator;
 
 import com.google.auto.service.AutoService;
 
@@ -48,15 +50,29 @@ public OptionRule optionRule() {
                 .optional(
                         MqttSourceOptions.USERNAME,
                         MqttSourceOptions.PASSWORD,
-                        MqttSourceOptions.QOS,
-                        MqttSourceOptions.FORMAT,
                         MqttSourceOptions.FIELD_DELIMITER,
                         MqttSourceOptions.CLIENT_ID,
                         MqttSourceOptions.CLEAN_SESSION,
                         MqttSourceOptions.CONNECTION_TIMEOUT,
-                        MqttSourceOptions.KEEP_ALIVE_INTERVAL,
+                        MqttSourceOptions.KEEP_ALIVE_INTERVAL)
+                .optional(
+                        MqttSourceOptions.QOS,
+                        Conditions.greaterOrEqual(MqttSourceOptions.QOS, 0)
+                                .and(Conditions.lessOrEqual(MqttSourceOptions.QOS, 1)))
+                .optional(
+                        MqttSourceOptions.FORMAT,
+                        Conditions.extension(MqttSourceOptions.FORMAT, new MqttFormatValidator()))
+                .optional(
                         MqttSourceOptions.RECONNECT_TIMEOUT,
-                        MqttSourceOptions.MAX_QUEUE_SIZE)
+                        Conditions.greaterThan(MqttSourceOptions.RECONNECT_TIMEOUT, 0))
+                .optional(
+                        MqttSourceOptions.MAX_QUEUE_SIZE,
+                        Conditions.greaterThan(MqttSourceOptions.MAX_QUEUE_SIZE, 0))
+                .conditional(MqttSourceOptions.CLEAN_SESSION, false, MqttSourceOptions.CLIENT_ID)
+                .conditional(
+                        MqttSourceOptions.CLEAN_SESSION,
+                        false,
+                        Conditions.notBlank(MqttSourceOptions.CLIENT_ID))
                 .build();
     }
 

seatunnel-connectors-v2/connector-mqtt/src/test/java/org/apache/seatunnel/connectors/seatunnel/mqtt/sink/MqttSinkFactoryTest.java

@@ -18,32 +18,117 @@
 package org.apache.seatunnel.connectors.seatunnel.mqtt.sink;
 
 import org.apache.seatunnel.api.configuration.Option;
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.configuration.util.ConfigValidator;
 import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.configuration.util.OptionValidationException;
 
 import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.stream.Collectors;
 
 public class MqttSinkFactoryTest {
 
+    private OptionRule sinkRule;
+
+    @BeforeEach
+    void setUp() {
+        sinkRule = new MqttSinkFactory().optionRule();
+    }
+
+    private static Map<String, Object> baseSinkConfig() {
+        Map<String, Object> cfg = new HashMap<>();
+        cfg.put(MqttSinkOptions.URL.key(), "tcp://localhost:1883");
+        cfg.put(MqttSinkOptions.TOPIC.key(), "test");
+        return cfg;
+    }
+
+    private void validateSink(Map<String, Object> cfg) {
+        ConfigValidator.of(ReadonlyConfig.fromMap(cfg)).validate(sinkRule);
+    }
+
     @Test
     void testOptionRule() {
-        MqttSinkFactory factory = new MqttSinkFactory();
-        OptionRule rule = factory.optionRule();
-
         List<Option<?>> requiredOptions =
-                rule.getRequiredOptions().stream()
+                sinkRule.getRequiredOptions().stream()
                         .flatMap(ro -> ro.getOptions().stream())
                         .collect(Collectors.toList());
         Assertions.assertTrue(requiredOptions.contains(MqttSinkOptions.URL));
         Assertions.assertTrue(requiredOptions.contains(MqttSinkOptions.TOPIC));
 
-        List<Option<?>> optionalOptions = rule.getOptionalOptions();
+        List<Option<?>> optionalOptions = sinkRule.getOptionalOptions();
         Assertions.assertTrue(optionalOptions.contains(MqttSinkOptions.QOS));
+        Assertions.assertTrue(optionalOptions.contains(MqttSinkOptions.FORMAT));
         Assertions.assertTrue(optionalOptions.contains(MqttSinkOptions.FIELD_DELIMITER));
         Assertions.assertTrue(optionalOptions.contains(MqttSinkOptions.BATCH_SIZE));
         Assertions.assertTrue(optionalOptions.contains(MqttSinkOptions.CLEAN_SESSION));
     }
+
+    @Test
+    void testValidBaseConfigPasses() {
+        Assertions.assertDoesNotThrow(() -> validateSink(baseSinkConfig()));
+    }
+
+    @Test
+    void testQosBoundariesPass() {
+        Map<String, Object> cfg = baseSinkConfig();
+        cfg.put(MqttSinkOptions.QOS.key(), 0);
+        Assertions.assertDoesNotThrow(() -> validateSink(cfg));
+
+        cfg.put(MqttSinkOptions.QOS.key(), 1);
+        Assertions.assertDoesNotThrow(() -> validateSink(cfg));
+    }
+
+    @Test
+    void testQosAboveRangeFails() {
+        Map<String, Object> cfg = baseSinkConfig();
+        cfg.put(MqttSinkOptions.QOS.key(), 2);
+        Assertions.assertThrows(OptionValidationException.class, () -> validateSink(cfg));
+    }
+
+    @Test
+    void testNegativeQosFails() {
+        Map<String, Object> cfg = baseSinkConfig();
+        cfg.put(MqttSinkOptions.QOS.key(), -1);
+        Assertions.assertThrows(OptionValidationException.class, () -> validateSink(cfg));
+    }
+
+    @Test
+    void testNonPositiveBatchSizeFails() {
+        Map<String, Object> cfg = baseSinkConfig();
+        cfg.put(MqttSinkOptions.BATCH_SIZE.key(), 0);
+        Assertions.assertThrows(OptionValidationException.class, () -> validateSink(cfg));
+
+        cfg.put(MqttSinkOptions.BATCH_SIZE.key(), -5);
+        Assertions.assertThrows(OptionValidationException.class, () -> validateSink(cfg));
+    }
+
+    @Test
+    void testMinimumBatchSizePasses() {
+        Map<String, Object> cfg = baseSinkConfig();
+        cfg.put(MqttSinkOptions.BATCH_SIZE.key(), 1);
+        Assertions.assertDoesNotThrow(() -> validateSink(cfg));
+    }
+
+    @Test
+    void testUnsupportedFormatFails() {
+        Map<String, Object> cfg = baseSinkConfig();
+        cfg.put(MqttSinkOptions.FORMAT.key(), "xml");
+        Assertions.assertThrows(OptionValidationException.class, () -> validateSink(cfg));
+    }
+
+    @Test
+    void testFormatIsCaseInsensitive() {
+        Map<String, Object> cfg = baseSinkConfig();
+        cfg.put(MqttSinkOptions.FORMAT.key(), "JSON");
+        Assertions.assertDoesNotThrow(() -> validateSink(cfg));
+
+        cfg.put(MqttSinkOptions.FORMAT.key(), "Text");
+        Assertions.assertDoesNotThrow(() -> validateSink(cfg));
+    }
 }

seatunnel-connectors-v2/connector-mqtt/src/test/java/org/apache/seatunnel/connectors/seatunnel/mqtt/sink/MqttSinkWriterTest.java

@@ -71,22 +71,8 @@ void setUp() {
         validConfig = ReadonlyConfig.fromMap(configMap);
     }
 
-    @Test
-    void testInvalidQosThrowsException() {
-        Map<String, Object> configMap = new HashMap<>();
-        configMap.put("url", "tcp://localhost:1883");
-        configMap.put("topic", "test");
-        configMap.put("qos", 2); // Invalid value
-
-        ReadonlyConfig config = ReadonlyConfig.fromMap(configMap);
-
-        IllegalArgumentException ex =
-                Assertions.assertThrows(
-                        IllegalArgumentException.class,
-                        () -> new MqttSinkWriter(context, rowType, config));
-
-        Assertions.assertTrue(ex.getMessage().contains("MQTT QoS must be 0"));
-    }
+    // Invalid qos and batch_size values are rejected declaratively by
+    // MqttSinkFactory#optionRule() before the writer is created; see MqttSinkFactoryTest.
 
     @Test
     void testInvalidFormatThrowsException() {