diff --git a/seatunnel-connectors-v2/connector-mqtt/src/main/java/org/apache/seatunnel/connectors/seatunnel/mqtt/config/MqttFormatValidator.java b/seatunnel-connectors-v2/connector-mqtt/src/main/java/org/apache/seatunnel/connectors/seatunnel/mqtt/config/MqttFormatValidator.java new file mode 100644 index 000000000000..3643421b37ca --- /dev/null +++ b/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 { + + @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); + } +} diff --git a/seatunnel-connectors-v2/connector-mqtt/src/main/java/org/apache/seatunnel/connectors/seatunnel/mqtt/sink/MqttSinkFactory.java b/seatunnel-connectors-v2/connector-mqtt/src/main/java/org/apache/seatunnel/connectors/seatunnel/mqtt/sink/MqttSinkFactory.java index 47e84069d799..9f42c1d058a5 100644 --- a/seatunnel-connectors-v2/connector-mqtt/src/main/java/org/apache/seatunnel/connectors/seatunnel/mqtt/sink/MqttSinkFactory.java +++ b/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(); } diff --git a/seatunnel-connectors-v2/connector-mqtt/src/main/java/org/apache/seatunnel/connectors/seatunnel/mqtt/sink/MqttSinkWriter.java b/seatunnel-connectors-v2/connector-mqtt/src/main/java/org/apache/seatunnel/connectors/seatunnel/mqtt/sink/MqttSinkWriter.java index c4e38e270211..17d540bd10a0 100644 --- a/seatunnel-connectors-v2/connector-mqtt/src/main/java/org/apache/seatunnel/connectors/seatunnel/mqtt/sink/MqttSinkWriter.java +++ b/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, 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); diff --git a/seatunnel-connectors-v2/connector-mqtt/src/main/java/org/apache/seatunnel/connectors/seatunnel/mqtt/source/MqttSourceConfig.java b/seatunnel-connectors-v2/connector-mqtt/src/main/java/org/apache/seatunnel/connectors/seatunnel/mqtt/source/MqttSourceConfig.java index e100e9bf3a36..aa68568a4f01 100644 --- a/seatunnel-connectors-v2/connector-mqtt/src/main/java/org/apache/seatunnel/connectors/seatunnel/mqtt/source/MqttSourceConfig.java +++ b/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) { diff --git a/seatunnel-connectors-v2/connector-mqtt/src/main/java/org/apache/seatunnel/connectors/seatunnel/mqtt/source/MqttSourceFactory.java b/seatunnel-connectors-v2/connector-mqtt/src/main/java/org/apache/seatunnel/connectors/seatunnel/mqtt/source/MqttSourceFactory.java index 8aca1f2bccbf..0640f90cdd19 100644 --- a/seatunnel-connectors-v2/connector-mqtt/src/main/java/org/apache/seatunnel/connectors/seatunnel/mqtt/source/MqttSourceFactory.java +++ b/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(); } diff --git a/seatunnel-connectors-v2/connector-mqtt/src/test/java/org/apache/seatunnel/connectors/seatunnel/mqtt/sink/MqttSinkFactoryTest.java b/seatunnel-connectors-v2/connector-mqtt/src/test/java/org/apache/seatunnel/connectors/seatunnel/mqtt/sink/MqttSinkFactoryTest.java index 29ed6a9a1dbd..863c7169ad69 100644 --- a/seatunnel-connectors-v2/connector-mqtt/src/test/java/org/apache/seatunnel/connectors/seatunnel/mqtt/sink/MqttSinkFactoryTest.java +++ b/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 baseSinkConfig() { + Map cfg = new HashMap<>(); + cfg.put(MqttSinkOptions.URL.key(), "tcp://localhost:1883"); + cfg.put(MqttSinkOptions.TOPIC.key(), "test"); + return cfg; + } + + private void validateSink(Map cfg) { + ConfigValidator.of(ReadonlyConfig.fromMap(cfg)).validate(sinkRule); + } + @Test void testOptionRule() { - MqttSinkFactory factory = new MqttSinkFactory(); - OptionRule rule = factory.optionRule(); - List> 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> optionalOptions = rule.getOptionalOptions(); + List> 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 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 cfg = baseSinkConfig(); + cfg.put(MqttSinkOptions.QOS.key(), 2); + Assertions.assertThrows(OptionValidationException.class, () -> validateSink(cfg)); + } + + @Test + void testNegativeQosFails() { + Map cfg = baseSinkConfig(); + cfg.put(MqttSinkOptions.QOS.key(), -1); + Assertions.assertThrows(OptionValidationException.class, () -> validateSink(cfg)); + } + + @Test + void testNonPositiveBatchSizeFails() { + Map 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 cfg = baseSinkConfig(); + cfg.put(MqttSinkOptions.BATCH_SIZE.key(), 1); + Assertions.assertDoesNotThrow(() -> validateSink(cfg)); + } + + @Test + void testUnsupportedFormatFails() { + Map cfg = baseSinkConfig(); + cfg.put(MqttSinkOptions.FORMAT.key(), "xml"); + Assertions.assertThrows(OptionValidationException.class, () -> validateSink(cfg)); + } + + @Test + void testFormatIsCaseInsensitive() { + Map cfg = baseSinkConfig(); + cfg.put(MqttSinkOptions.FORMAT.key(), "JSON"); + Assertions.assertDoesNotThrow(() -> validateSink(cfg)); + + cfg.put(MqttSinkOptions.FORMAT.key(), "Text"); + Assertions.assertDoesNotThrow(() -> validateSink(cfg)); + } } diff --git a/seatunnel-connectors-v2/connector-mqtt/src/test/java/org/apache/seatunnel/connectors/seatunnel/mqtt/sink/MqttSinkWriterTest.java b/seatunnel-connectors-v2/connector-mqtt/src/test/java/org/apache/seatunnel/connectors/seatunnel/mqtt/sink/MqttSinkWriterTest.java index 263beb0adab0..7f863ed9a971 100644 --- a/seatunnel-connectors-v2/connector-mqtt/src/test/java/org/apache/seatunnel/connectors/seatunnel/mqtt/sink/MqttSinkWriterTest.java +++ b/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 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() { diff --git a/seatunnel-connectors-v2/connector-mqtt/src/test/java/org/apache/seatunnel/connectors/seatunnel/mqtt/source/MqttSourceConfigTest.java b/seatunnel-connectors-v2/connector-mqtt/src/test/java/org/apache/seatunnel/connectors/seatunnel/mqtt/source/MqttSourceConfigTest.java index f9c44977363e..ce277a2136be 100644 --- a/seatunnel-connectors-v2/connector-mqtt/src/test/java/org/apache/seatunnel/connectors/seatunnel/mqtt/source/MqttSourceConfigTest.java +++ b/seatunnel-connectors-v2/connector-mqtt/src/test/java/org/apache/seatunnel/connectors/seatunnel/mqtt/source/MqttSourceConfigTest.java @@ -75,76 +75,9 @@ void testExplicitValues() { Assertions.assertEquals(200, sourceConfig.getMaxQueueSize()); } - @Test - void testInvalidQosFails() { - Map config = baseConfig(); - config.put("qos", 2); - - Assertions.assertThrows( - IllegalArgumentException.class, - () -> new MqttSourceConfig(ReadonlyConfig.fromMap(config))); - } - - @Test - void testNegativeQosFails() { - Map config = baseConfig(); - config.put("qos", -1); - - Assertions.assertThrows( - IllegalArgumentException.class, - () -> new MqttSourceConfig(ReadonlyConfig.fromMap(config))); - } - - @Test - void testNonPositiveReconnectTimeoutFails() { - Map config = baseConfig(); - config.put("reconnect_timeout", 0); - - Assertions.assertThrows( - IllegalArgumentException.class, - () -> new MqttSourceConfig(ReadonlyConfig.fromMap(config))); - } - - @Test - void testNonPositiveMaxQueueSizeFails() { - Map config = baseConfig(); - config.put("max_queue_size", 0); - - Assertions.assertThrows( - IllegalArgumentException.class, - () -> new MqttSourceConfig(ReadonlyConfig.fromMap(config))); - } - - @Test - void testUnsupportedFormatFails() { - Map config = baseConfig(); - config.put("format", "avro"); - - Assertions.assertThrows( - IllegalArgumentException.class, - () -> new MqttSourceConfig(ReadonlyConfig.fromMap(config))); - } - - @Test - void testPersistentSessionRequiresClientId() { - Map config = baseConfig(); - config.put("clean_session", false); - - Assertions.assertThrows( - IllegalArgumentException.class, - () -> new MqttSourceConfig(ReadonlyConfig.fromMap(config))); - } - - @Test - void testPersistentSessionRequiresNonBlankClientId() { - Map config = baseConfig(); - config.put("clean_session", false); - config.put("client_id", " "); - - Assertions.assertThrows( - IllegalArgumentException.class, - () -> new MqttSourceConfig(ReadonlyConfig.fromMap(config))); - } + // Invalid option values (qos range, format, reconnect_timeout, max_queue_size, missing or + // blank client_id for persistent sessions) are rejected declaratively by + // MqttSourceFactory#optionRule(); see MqttSourceFactoryTest. @Test void testDefaultClientIdIsGeneratedForCleanSession() { diff --git a/seatunnel-connectors-v2/connector-mqtt/src/test/java/org/apache/seatunnel/connectors/seatunnel/mqtt/source/MqttSourceFactoryTest.java b/seatunnel-connectors-v2/connector-mqtt/src/test/java/org/apache/seatunnel/connectors/seatunnel/mqtt/source/MqttSourceFactoryTest.java index 210376bd5614..2a9a5b65cc03 100644 --- a/seatunnel-connectors-v2/connector-mqtt/src/test/java/org/apache/seatunnel/connectors/seatunnel/mqtt/source/MqttSourceFactoryTest.java +++ b/seatunnel-connectors-v2/connector-mqtt/src/test/java/org/apache/seatunnel/connectors/seatunnel/mqtt/source/MqttSourceFactoryTest.java @@ -18,17 +18,45 @@ package org.apache.seatunnel.connectors.seatunnel.mqtt.source; 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.apache.seatunnel.api.options.ConnectorCommonOptions; import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import java.util.Collections; +import java.util.HashMap; import java.util.List; +import java.util.Map; import java.util.stream.Collectors; public class MqttSourceFactoryTest { + private OptionRule sourceRule; + + @BeforeEach + void setUp() { + sourceRule = new MqttSourceFactory().optionRule(); + } + + private static Map baseSourceConfig() { + Map cfg = new HashMap<>(); + cfg.put(MqttSourceOptions.URL.key(), "tcp://localhost:1883"); + cfg.put(MqttSourceOptions.TOPIC.key(), "users"); + cfg.put( + ConnectorCommonOptions.SCHEMA.key(), + Collections.singletonMap("fields", Collections.singletonMap("name", "string"))); + return cfg; + } + + private void validateSource(Map cfg) { + ConfigValidator.of(ReadonlyConfig.fromMap(cfg)).validate(sourceRule); + } + @Test void testFactoryIdentifier() { MqttSourceFactory factory = new MqttSourceFactory(); @@ -39,18 +67,15 @@ void testFactoryIdentifier() { @Test void testOptionRule() { - MqttSourceFactory factory = new MqttSourceFactory(); - OptionRule rule = factory.optionRule(); - List> requiredOptions = - rule.getRequiredOptions().stream() + sourceRule.getRequiredOptions().stream() .flatMap(requiredOption -> requiredOption.getOptions().stream()) .collect(Collectors.toList()); Assertions.assertTrue(requiredOptions.contains(MqttSourceOptions.URL)); Assertions.assertTrue(requiredOptions.contains(MqttSourceOptions.TOPIC)); Assertions.assertTrue(requiredOptions.contains(ConnectorCommonOptions.SCHEMA)); - List> optionalOptions = rule.getOptionalOptions(); + List> optionalOptions = sourceRule.getOptionalOptions(); Assertions.assertTrue(optionalOptions.contains(MqttSourceOptions.USERNAME)); Assertions.assertTrue(optionalOptions.contains(MqttSourceOptions.PASSWORD)); Assertions.assertTrue(optionalOptions.contains(MqttSourceOptions.QOS)); @@ -60,6 +85,90 @@ void testOptionRule() { Assertions.assertTrue(optionalOptions.contains(MqttSourceOptions.CLEAN_SESSION)); Assertions.assertTrue(optionalOptions.contains(MqttSourceOptions.CONNECTION_TIMEOUT)); Assertions.assertTrue(optionalOptions.contains(MqttSourceOptions.KEEP_ALIVE_INTERVAL)); + Assertions.assertTrue(optionalOptions.contains(MqttSourceOptions.RECONNECT_TIMEOUT)); Assertions.assertTrue(optionalOptions.contains(MqttSourceOptions.MAX_QUEUE_SIZE)); } + + @Test + void testValidBaseConfigPasses() { + Assertions.assertDoesNotThrow(() -> validateSource(baseSourceConfig())); + } + + @Test + void testQosBoundariesPass() { + Map cfg = baseSourceConfig(); + cfg.put(MqttSourceOptions.QOS.key(), 0); + Assertions.assertDoesNotThrow(() -> validateSource(cfg)); + + cfg.put(MqttSourceOptions.QOS.key(), 1); + Assertions.assertDoesNotThrow(() -> validateSource(cfg)); + } + + @Test + void testQosAboveRangeFails() { + Map cfg = baseSourceConfig(); + cfg.put(MqttSourceOptions.QOS.key(), 2); + Assertions.assertThrows(OptionValidationException.class, () -> validateSource(cfg)); + } + + @Test + void testNegativeQosFails() { + Map cfg = baseSourceConfig(); + cfg.put(MqttSourceOptions.QOS.key(), -1); + Assertions.assertThrows(OptionValidationException.class, () -> validateSource(cfg)); + } + + @Test + void testNonPositiveReconnectTimeoutFails() { + Map cfg = baseSourceConfig(); + cfg.put(MqttSourceOptions.RECONNECT_TIMEOUT.key(), 0); + Assertions.assertThrows(OptionValidationException.class, () -> validateSource(cfg)); + } + + @Test + void testNonPositiveMaxQueueSizeFails() { + Map cfg = baseSourceConfig(); + cfg.put(MqttSourceOptions.MAX_QUEUE_SIZE.key(), 0); + Assertions.assertThrows(OptionValidationException.class, () -> validateSource(cfg)); + } + + @Test + void testUnsupportedFormatFails() { + Map cfg = baseSourceConfig(); + cfg.put(MqttSourceOptions.FORMAT.key(), "avro"); + Assertions.assertThrows(OptionValidationException.class, () -> validateSource(cfg)); + } + + @Test + void testFormatIsCaseInsensitive() { + Map cfg = baseSourceConfig(); + cfg.put(MqttSourceOptions.FORMAT.key(), "JSON"); + Assertions.assertDoesNotThrow(() -> validateSource(cfg)); + + cfg.put(MqttSourceOptions.FORMAT.key(), "Text"); + Assertions.assertDoesNotThrow(() -> validateSource(cfg)); + } + + @Test + void testPersistentSessionRequiresClientId() { + Map cfg = baseSourceConfig(); + cfg.put(MqttSourceOptions.CLEAN_SESSION.key(), false); + Assertions.assertThrows(OptionValidationException.class, () -> validateSource(cfg)); + } + + @Test + void testPersistentSessionRequiresNonBlankClientId() { + Map cfg = baseSourceConfig(); + cfg.put(MqttSourceOptions.CLEAN_SESSION.key(), false); + cfg.put(MqttSourceOptions.CLIENT_ID.key(), " "); + Assertions.assertThrows(OptionValidationException.class, () -> validateSource(cfg)); + } + + @Test + void testPersistentSessionWithClientIdPasses() { + Map cfg = baseSourceConfig(); + cfg.put(MqttSourceOptions.CLEAN_SESSION.key(), false); + cfg.put(MqttSourceOptions.CLIENT_ID.key(), "mqtt-source-client"); + Assertions.assertDoesNotThrow(() -> validateSource(cfg)); + } }