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 801e330cba [Test][Connector-V2] Complete the MQTT serialization field
coverage (#12416)
801e330cba is described below
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() {
