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-12410-36c0d1b8a6fc880f208384268f7bec435e7bf601 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit d710e0678d0357ac3f22ffb694f32cb21da36ac6 Author: Sepuri Sai Krishna <[email protected]> AuthorDate: Mon Sep 21 06:04:51 2026 +0000 [Test][Connector-V2] Strengthen the MQTT source serialization coverage (#12410) --- .../mqtt/source/MqttSourceConfigTest.java | 34 ++++++++++++++++++++++ .../seatunnel/mqtt/source/MqttSourceTest.java | 14 ++++++++- 2 files changed, 47 insertions(+), 1 deletion(-) 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 f9c4497736..0582f77cd1 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 @@ -22,6 +22,10 @@ import org.apache.seatunnel.api.configuration.ReadonlyConfig; 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; @@ -153,6 +157,36 @@ class MqttSourceConfigTest { Assertions.assertTrue(sourceConfig.getClientId().startsWith("seatunnel_mqtt_source_")); } + @Test + void testConfigSurvivesJavaSerialization() throws Exception { + Map<String, Object> config = baseConfig(); + 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("format", "text"); + 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()); + } + } + private static Map<String, Object> baseConfig() { Map<String, Object> config = new HashMap<>(); config.put("url", "tcp://localhost:1883"); 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 0e027e0f5f..72cd2caa62 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 @@ -204,6 +204,9 @@ class MqttSourceTest { @Test void testSourceIsJavaSerializable() throws Exception { 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(); @@ -215,7 +218,16 @@ class MqttSourceTest { try (ObjectInputStream in = new ObjectInputStream(new ByteArrayInputStream(serialized))) { MqttSource restored = (MqttSource) in.readObject(); - Assertions.assertEquals(source.getPluginName(), restored.getPluginName()); + // 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()); } }
