This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12416-7f7e9a7fc33345ac4e788de908b1893b4949a089 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 801e330cbaf69345e644300f8cf00bb8878b4798 Author: Sepuri Sai Krishna <[email protected]> AuthorDate: Tue Sep 22 04:29:30 2026 +0000 [Test][Connector-V2] Complete the MQTT serialization field coverage (#12416) --- .../mqtt/source/MqttSourceConfigTest.java | 64 +++++++++++++--------- .../seatunnel/mqtt/source/MqttSourceTest.java | 56 ++++++++++--------- 2 files changed, 69 insertions(+), 51 deletions(-) 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 0582f77cd1..e8ad05516c 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 @@ -18,14 +18,11 @@ package org.apache.seatunnel.connectors.seatunnel.mqtt.source; import org.apache.seatunnel.api.configuration.ReadonlyConfig; +import org.apache.seatunnel.common.utils.SerializationUtils; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; -import java.io.ByteArrayInputStream; -import java.io.ByteArrayOutputStream; -import java.io.ObjectInputStream; -import java.io.ObjectOutputStream; import java.util.HashMap; import java.util.Map; @@ -158,33 +155,48 @@ class MqttSourceConfigTest { } @Test - void testConfigSurvivesJavaSerialization() throws Exception { + void testConfigSurvivesJavaSerialization() { Map<String, Object> config = baseConfig(); + // Deserialization does not run the constructor, so the failure this guards against is + // a field that stops being written at all, which leaves the type's zero value behind. + // Every value below is therefore chosen to differ from that zero value: null for the + // references, 0 for the ints, false for the boolean. qos and clean_session are the two + // that constrain the choice, since qos is limited to 0 or 1 and 0 is the zero value, + // so each has to take the value that is not zero even though it matches the default. config.put("username", "admin"); - // Deliberately non-default values, so the assertions below prove the values round - // tripped rather than that defaults were rebuilt. The default qos is 1 and validate() - // limits it to 0 or 1, so 0 is the only non-default that is legal; the default format - // is json. - config.put("qos", 0); + config.put("password", "s3cret"); + config.put("qos", 1); config.put("format", "text"); + config.put("field_delimiter", "|"); + config.put("clean_session", true); + config.put("client_id", "mqtt-source-round-trip"); + config.put("connection_timeout", 45); + config.put("keep_alive_interval", 90); + config.put("reconnect_timeout", 150); + config.put("max_queue_size", 2048); MqttSourceConfig sourceConfig = new MqttSourceConfig(ReadonlyConfig.fromMap(config)); - byte[] serialized; - try (ByteArrayOutputStream bytes = new ByteArrayOutputStream(); - ObjectOutputStream out = new ObjectOutputStream(bytes)) { - out.writeObject(sourceConfig); - out.flush(); - serialized = bytes.toByteArray(); - } - - try (ObjectInputStream in = new ObjectInputStream(new ByteArrayInputStream(serialized))) { - MqttSourceConfig restored = (MqttSourceConfig) in.readObject(); - Assertions.assertEquals(sourceConfig.getUrl(), restored.getUrl()); - Assertions.assertEquals(sourceConfig.getTopic(), restored.getTopic()); - Assertions.assertEquals(sourceConfig.getUsername(), restored.getUsername()); - Assertions.assertEquals(sourceConfig.getQos(), restored.getQos()); - Assertions.assertEquals(sourceConfig.getClientId(), restored.getClientId()); - } + MqttSourceConfig restored = + SerializationUtils.deserialize(SerializationUtils.serialize(sourceConfig)); + + // Assert every field, not a sample. MqttSourceReader reads password, format, + // fieldDelimiter, the timeouts and maxQueueSize on the worker, so marking any one of + // them transient has to fail here rather than against a live broker. + Assertions.assertEquals(sourceConfig.getUrl(), restored.getUrl()); + Assertions.assertEquals(sourceConfig.getTopic(), restored.getTopic()); + Assertions.assertEquals(sourceConfig.getUsername(), restored.getUsername()); + Assertions.assertEquals(sourceConfig.getPassword(), restored.getPassword()); + Assertions.assertEquals(sourceConfig.getQos(), restored.getQos()); + Assertions.assertEquals(sourceConfig.getFormat(), restored.getFormat()); + Assertions.assertEquals(sourceConfig.getFieldDelimiter(), restored.getFieldDelimiter()); + Assertions.assertEquals(sourceConfig.getClientId(), restored.getClientId()); + Assertions.assertEquals(sourceConfig.isCleanSession(), restored.isCleanSession()); + Assertions.assertEquals( + sourceConfig.getConnectionTimeout(), restored.getConnectionTimeout()); + Assertions.assertEquals( + sourceConfig.getKeepAliveInterval(), restored.getKeepAliveInterval()); + Assertions.assertEquals(sourceConfig.getReconnectTimeout(), restored.getReconnectTimeout()); + Assertions.assertEquals(sourceConfig.getMaxQueueSize(), restored.getMaxQueueSize()); } private static Map<String, Object> baseConfig() { diff --git a/seatunnel-connectors-v2/connector-mqtt/src/test/java/org/apache/seatunnel/connectors/seatunnel/mqtt/source/MqttSourceTest.java b/seatunnel-connectors-v2/connector-mqtt/src/test/java/org/apache/seatunnel/connectors/seatunnel/mqtt/source/MqttSourceTest.java index 72cd2caa62..13dd330d7c 100644 --- a/seatunnel-connectors-v2/connector-mqtt/src/test/java/org/apache/seatunnel/connectors/seatunnel/mqtt/source/MqttSourceTest.java +++ b/seatunnel-connectors-v2/connector-mqtt/src/test/java/org/apache/seatunnel/connectors/seatunnel/mqtt/source/MqttSourceTest.java @@ -24,6 +24,7 @@ import org.apache.seatunnel.api.source.Collector; import org.apache.seatunnel.api.table.catalog.CatalogTable; import org.apache.seatunnel.api.table.type.SeaTunnelRow; import org.apache.seatunnel.common.constants.JobMode; +import org.apache.seatunnel.common.utils.SerializationUtils; import org.apache.seatunnel.connectors.seatunnel.common.source.AbstractSingleSplitReader; import org.apache.seatunnel.connectors.seatunnel.common.source.SingleSplitReaderContext; import org.apache.seatunnel.connectors.seatunnel.mqtt.exception.MqttConnectorException; @@ -35,10 +36,6 @@ import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; import org.mockito.Mockito; -import java.io.ByteArrayInputStream; -import java.io.ByteArrayOutputStream; -import java.io.ObjectInputStream; -import java.io.ObjectOutputStream; import java.nio.charset.StandardCharsets; import java.util.Collections; import java.util.HashMap; @@ -202,33 +199,42 @@ class MqttSourceTest { } @Test - void testSourceIsJavaSerializable() throws Exception { + void testSourceIsJavaSerializable() { MqttSource source = new MqttSource(ReadonlyConfig.fromMap(baseConfig())); // The engine sets the job context before it serializes the logical DAG, so mirror that // ordering rather than serializing a source that has never had one. source.setJobContext(new JobContext().setJobMode(JobMode.STREAMING)); - byte[] serialized; - try (ByteArrayOutputStream bytes = new ByteArrayOutputStream(); - ObjectOutputStream out = new ObjectOutputStream(bytes)) { - out.writeObject(source); - out.flush(); - serialized = bytes.toByteArray(); - } + MqttSource restored = SerializationUtils.deserialize(SerializationUtils.serialize(source)); - try (ObjectInputStream in = new ObjectInputStream(new ByteArrayInputStream(serialized))) { - MqttSource restored = (MqttSource) in.readObject(); - // Assert restored state, not a constant: getPluginName() alone would still pass if - // every field were lost. The catalog table and the job context are the state the - // engine relies on after deserializing the vertex. - Assertions.assertEquals( - source.getProducedCatalogTables().get(0).getTableId(), - restored.getProducedCatalogTables().get(0).getTableId()); - Assertions.assertEquals( - source.getProducedCatalogTables().get(0).getTableSchema(), - restored.getProducedCatalogTables().get(0).getTableSchema()); - Assertions.assertEquals(Boundedness.UNBOUNDED, restored.getBoundedness()); - } + // Assert restored state, not a constant: getPluginName() alone would still pass if + // every field were lost. + Assertions.assertEquals( + source.getProducedCatalogTables().get(0).getTableId(), + restored.getProducedCatalogTables().get(0).getTableId()); + Assertions.assertEquals( + source.getProducedCatalogTables().get(0).getTableSchema(), + restored.getProducedCatalogTables().get(0).getTableSchema()); + + // sourceConfig is the third and last field of MqttSource, and the one createReader + // hands to the reader. There is no getter for it, but MqttSourceReader's constructor + // dereferences it, so building a reader from the restored source fails if it was lost. + Assertions.assertDoesNotThrow( + () -> restored.createReader(Mockito.mock(SingleSplitReaderContext.class))); + } + + @Test + void testJobContextSurvivesJavaSerialization() { + MqttSource source = new MqttSource(ReadonlyConfig.fromMap(baseConfig())); + // A STREAMING round trip cannot prove the job context survived: getBoundedness() + // returns UNBOUNDED for a null context too, so the assertion would hold even if the + // field were dropped. BATCH distinguishes the two, because it only throws when the + // context came back. The pre-serialization equivalent is testBatchJobModeFails. + source.setJobContext(new JobContext().setJobMode(JobMode.BATCH)); + + MqttSource restored = SerializationUtils.deserialize(SerializationUtils.serialize(source)); + + Assertions.assertThrows(MqttConnectorException.class, restored::getBoundedness); } private static Map<String, Object> baseConfig() {
