Skip to content

Commit 4fd1923

Browse files
assokhiassokhi
committed
trigger github actions
1 parent c74124a commit 4fd1923

10 files changed

Lines changed: 620 additions & 99 deletions

File tree

‎docs/en/connectors/changelog/connector-mqtt.md‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,4 +4,5 @@
44

55
### Sink
66

7-
- Add MQTT Sink Connector ([#XXXX](https://github.com/apache/seatunnel/pull/XXXX))
7+
- Add MQTT Sink Connector ([#10575](https://github.com/apache/seatunnel/pull/10575))
8+
Resolves [#9566](https://github.com/apache/seatunnel/issues/9566)

‎docs/en/connectors/sink/Mqtt.md‎

Lines changed: 42 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,11 @@ This connector is suitable for publishing SeaTunnel pipeline data to IoT endpoin
1414

1515
- [ ] [exactly-once](../../introduction/concepts/connector-v2-features.md)
1616

17-
MQTT is a stateless publish/subscribe protocol without distributed transaction support. The connector provides **at-least-once** delivery semantics by relying on SeaTunnel's source-replay mechanics and MQTT QoS 1.
17+
**Delivery Semantics Notice**:
18+
This connector provides **at-most-once** delivery when QoS=0, and **best-effort at-least-once** when QoS=1.
19+
Due to `clean_session=true` (the default, required for stateless operation), unacknowledged messages may be lost during
20+
client disconnections. For stronger guarantees, consider setting `clean_session=false` (with proper clientId management)
21+
or enabling source replay capabilities in SeaTunnel.
1822

1923
## Supported Engines
2024

@@ -32,8 +36,11 @@ MQTT is a stateless publish/subscribe protocol without distributed transaction s
3236
| password | string | no | - |
3337
| qos | int | no | 1 |
3438
| format | string | no | json |
39+
| field_delimiter | string | no | , |
40+
| batch_size | int | no | 1 |
3541
| retry_timeout | int | no | 5000 |
3642
| connection_timeout | int | no | 30 |
43+
| clean_session | boolean | no | true |
3744
| common-options | | no | - |
3845

3946
### url [string]
@@ -68,7 +75,19 @@ The MQTT Quality of Service level for published messages.
6875
The serialization format for outgoing messages. Supported values:
6976

7077
- `json` — Serialize each row as a JSON object (default)
71-
- `text` — Serialize each row as comma-delimited plain text
78+
- `text` — Serialize each row as delimited plain text (delimiter controlled by `field_delimiter`)
79+
80+
### field_delimiter [string]
81+
82+
The field delimiter used when `format` is set to `text`. Default is `,`.
83+
84+
Examples: `,`, `|`, `\t`
85+
86+
### batch_size [int]
87+
88+
Number of messages to buffer before sending to the broker. Default is `1` (send each message immediately).
89+
90+
Higher values improve throughput by reducing per-message overhead. Buffered messages are automatically flushed at each checkpoint and when the writer closes.
7291

7392
### retry_timeout [int]
7493

@@ -78,10 +97,31 @@ Maximum time in milliseconds to retry publishing on transient network failures b
7897

7998
The MQTT connection establishment timeout in seconds.
8099

100+
### clean_session [boolean]
101+
102+
Whether to use a clean MQTT session. Default is `true`.
103+
104+
- `true` — Broker discards any previous session state. Suitable for stateless operation (recommended for most use cases).
105+
- `false` — Broker retains session state (subscriptions, unacknowledged QoS 1 messages). Enables stronger at-least-once guarantees but may cause broker-side state accumulation. Requires stable, unique `clientId` per writer.
106+
81107
### common options
82108

83109
Sink plugin common parameters, please refer to [Sink Common Options](../common-options/sink-common-options.md) for details.
84110

111+
## Performance Considerations
112+
113+
The MQTT Sink sends messages synchronously to guarantee delivery ordering. Typical throughput:
114+
115+
- QoS 0: ~10,000 messages/sec (local network)
116+
- QoS 1: ~5,000 messages/sec (requires broker ACK)
117+
118+
To improve throughput:
119+
120+
- Increase `batch_size` to reduce per-message overhead (e.g., `batch_size = 100`)
121+
- Reduce `qos` to `0` if at-most-once delivery is acceptable
122+
- Increase SeaTunnel parallelism to distribute load across multiple MQTT clients
123+
- For very high throughput requirements, consider using the Kafka Sink instead
124+
85125
## Example
86126

87127
### Simple JSON sink

‎seatunnel-connectors-v2/connector-mqtt/src/main/java/org/apache/seatunnel/connectors/seatunnel/mqtt/exception/MqttConnectorErrorCode.java‎

Lines changed: 44 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,44 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.seatunnel.connectors.seatunnel.mqtt.exception;
19+
20+
import org.apache.seatunnel.common.exception.SeaTunnelErrorCode;
21+
22+
public enum MqttConnectorErrorCode implements SeaTunnelErrorCode {
23+
CONNECTION_FAILED("MQTT-01", "MQTT connection failed"),
24+
PUBLISH_FAILED("MQTT-02", "MQTT message publish failed"),
25+
INVALID_CONFIG("MQTT-03", "Invalid MQTT configuration");
26+
27+
private final String code;
28+
private final String description;
29+
30+
MqttConnectorErrorCode(String code, String description) {
31+
this.code = code;
32+
this.description = description;
33+
}
34+
35+
@Override
36+
public String getCode() {
37+
return code;
38+
}
39+
40+
@Override
41+
public String getDescription() {
42+
return description;
43+
}
44+
}

‎seatunnel-connectors-v2/connector-mqtt/src/main/java/org/apache/seatunnel/connectors/seatunnel/mqtt/exception/MqttConnectorException.java‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,32 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.seatunnel.connectors.seatunnel.mqtt.exception;
19+
20+
import org.apache.seatunnel.common.exception.SeaTunnelErrorCode;
21+
import org.apache.seatunnel.common.exception.SeaTunnelRuntimeException;
22+
23+
public class MqttConnectorException extends SeaTunnelRuntimeException {
24+
public MqttConnectorException(SeaTunnelErrorCode errorCode, String errorMessage) {
25+
super(errorCode, errorMessage);
26+
}
27+
28+
public MqttConnectorException(
29+
SeaTunnelErrorCode errorCode, String errorMessage, Throwable cause) {
30+
super(errorCode, errorMessage, cause);
31+
}
32+
}

‎seatunnel-connectors-v2/connector-mqtt/src/main/java/org/apache/seatunnel/connectors/seatunnel/mqtt/sink/MqttSinkFactory.java‎

Lines changed: 10 additions & 53 deletions
Original file line numberDiff line numberDiff line change
@@ -17,8 +17,6 @@
1717

1818
package org.apache.seatunnel.connectors.seatunnel.mqtt.sink;
1919

20-
import org.apache.seatunnel.api.configuration.Option;
21-
import org.apache.seatunnel.api.configuration.Options;
2220
import org.apache.seatunnel.api.configuration.util.OptionRule;
2321
import org.apache.seatunnel.api.table.connector.TableSink;
2422
import org.apache.seatunnel.api.table.factory.Factory;
@@ -30,55 +28,6 @@
3028
@AutoService(Factory.class)
3129
public class MqttSinkFactory implements TableSinkFactory {
3230

33-
public static final Option<String> URL =
34-
Options.key("url")
35-
.stringType()
36-
.noDefaultValue()
37-
.withDescription("MQTT broker URL, e.g. tcp://localhost:1883");
38-
39-
public static final Option<String> TOPIC =
40-
Options.key("topic")
41-
.stringType()
42-
.noDefaultValue()
43-
.withDescription("Target MQTT topic to publish messages to");
44-
45-
public static final Option<String> USERNAME =
46-
Options.key("username")
47-
.stringType()
48-
.noDefaultValue()
49-
.withDescription("MQTT broker authentication username");
50-
51-
public static final Option<String> PASSWORD =
52-
Options.key("password")
53-
.stringType()
54-
.noDefaultValue()
55-
.withDescription("MQTT broker authentication password");
56-
57-
public static final Option<Integer> QOS =
58-
Options.key("qos")
59-
.intType()
60-
.defaultValue(1)
61-
.withDescription("MQTT QoS level: 0 (at-most-once), 1 (at-least-once)");
62-
63-
public static final Option<String> FORMAT =
64-
Options.key("format")
65-
.stringType()
66-
.defaultValue("json")
67-
.withDescription("Message serialization format: json or text");
68-
69-
public static final Option<Integer> RETRY_TIMEOUT =
70-
Options.key("retry_timeout")
71-
.intType()
72-
.defaultValue(5000)
73-
.withDescription(
74-
"Maximum time in milliseconds to retry publishing on transient failures");
75-
76-
public static final Option<Integer> CONNECTION_TIMEOUT =
77-
Options.key("connection_timeout")
78-
.intType()
79-
.defaultValue(30)
80-
.withDescription("MQTT connection timeout in seconds");
81-
8231
@Override
8332
public String factoryIdentifier() {
8433
return "MQTT";
@@ -87,8 +36,16 @@ public String factoryIdentifier() {
8736
@Override
8837
public OptionRule optionRule() {
8938
return OptionRule.builder()
90-
.required(URL, TOPIC)
91-
.optional(USERNAME, PASSWORD, QOS, FORMAT, RETRY_TIMEOUT, CONNECTION_TIMEOUT)
39+
.required(MqttSinkOptions.URL, MqttSinkOptions.TOPIC)
40+
.optional(
41+
MqttSinkOptions.USERNAME,
42+
MqttSinkOptions.PASSWORD,
43+
MqttSinkOptions.QOS,
44+
MqttSinkOptions.FORMAT,
45+
MqttSinkOptions.FIELD_DELIMITER,
46+
MqttSinkOptions.BATCH_SIZE,
47+
MqttSinkOptions.RETRY_TIMEOUT,
48+
MqttSinkOptions.CONNECTION_TIMEOUT)
9249
.build();
9350
}
9451

‎seatunnel-connectors-v2/connector-mqtt/src/main/java/org/apache/seatunnel/connectors/seatunnel/mqtt/sink/MqttSinkOptions.java‎

Lines changed: 95 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,95 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.seatunnel.connectors.seatunnel.mqtt.sink;
19+
20+
import org.apache.seatunnel.api.configuration.Option;
21+
import org.apache.seatunnel.api.configuration.Options;
22+
23+
public class MqttSinkOptions {
24+
25+
public static final Option<String> URL =
26+
Options.key("url")
27+
.stringType()
28+
.noDefaultValue()
29+
.withDescription("MQTT broker URL, e.g. tcp://localhost:1883");
30+
31+
public static final Option<String> TOPIC =
32+
Options.key("topic")
33+
.stringType()
34+
.noDefaultValue()
35+
.withDescription("Target MQTT topic to publish messages to");
36+
37+
public static final Option<String> USERNAME =
38+
Options.key("username")
39+
.stringType()
40+
.noDefaultValue()
41+
.withDescription("MQTT broker authentication username");
42+
43+
public static final Option<String> PASSWORD =
44+
Options.key("password")
45+
.stringType()
46+
.noDefaultValue()
47+
.withDescription("MQTT broker authentication password");
48+
49+
public static final Option<Integer> QOS =
50+
Options.key("qos")
51+
.intType()
52+
.defaultValue(1)
53+
.withDescription("MQTT QoS level: 0 (at-most-once), 1 (at-least-once)");
54+
55+
public static final Option<String> FORMAT =
56+
Options.key("format")
57+
.stringType()
58+
.defaultValue("json")
59+
.withDescription("Message serialization format: json or text");
60+
61+
public static final Option<String> FIELD_DELIMITER =
62+
Options.key("field_delimiter")
63+
.stringType()
64+
.defaultValue(",")
65+
.withDescription("Field delimiter for text format. Only used when format=text");
66+
67+
public static final Option<Integer> BATCH_SIZE =
68+
Options.key("batch_size")
69+
.intType()
70+
.defaultValue(1)
71+
.withDescription(
72+
"Number of messages to buffer before sending. "
73+
+ "Higher values improve throughput by reducing per-message overhead. "
74+
+ "Buffered messages are also flushed at each checkpoint.");
75+
76+
public static final Option<Integer> RETRY_TIMEOUT =
77+
Options.key("retry_timeout")
78+
.intType()
79+
.defaultValue(5000)
80+
.withDescription(
81+
"Maximum time in milliseconds to retry publishing on transient failures");
82+
83+
public static final Option<Integer> CONNECTION_TIMEOUT =
84+
Options.key("connection_timeout")
85+
.intType()
86+
.defaultValue(30)
87+
.withDescription("MQTT connection timeout in seconds");
88+
89+
public static final Option<Boolean> CLEAN_SESSION =
90+
Options.key("clean_session")
91+
.booleanType()
92+
.defaultValue(true)
93+
.withDescription(
94+
"Whether to use clean session. false enables persistent sessions but may cause broker-side state accumulation.");
95+
}

0 commit comments

Comments
 (0)