This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new d710e0678d [Test][Connector-V2] Strengthen the MQTT source
serialization coverage (#12410)
d710e0678d is described below
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());
}
}
