Skip to content

Commit 41ff5b5

Browse files
assokhiassokhi
committed
trigger github actions
1 parent 249685f commit 41ff5b5

4 files changed

Lines changed: 106 additions & 62 deletions

File tree

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

Lines changed: 8 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,14 @@ 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.RETRY_TIMEOUT,
46+
MqttSinkOptions.CONNECTION_TIMEOUT)
9247
.build();
9348
}
9449

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

Lines changed: 73 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,73 @@
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<Integer> RETRY_TIMEOUT =
62+
Options.key("retry_timeout")
63+
.intType()
64+
.defaultValue(5000)
65+
.withDescription(
66+
"Maximum time in milliseconds to retry publishing on transient failures");
67+
68+
public static final Option<Integer> CONNECTION_TIMEOUT =
69+
Options.key("connection_timeout")
70+
.intType()
71+
.defaultValue(30)
72+
.withDescription("MQTT connection timeout in seconds");
73+
}

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

Lines changed: 9 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -57,9 +57,9 @@ public class MqttSinkWriter implements SinkWriter<SeaTunnelRow, Void, Void>, Mqt
5757

5858
public MqttSinkWriter(
5959
SinkWriter.Context context, SeaTunnelRowType rowType, ReadonlyConfig pluginConfig) {
60-
this.topic = pluginConfig.get(MqttSinkFactory.TOPIC);
61-
this.qos = pluginConfig.get(MqttSinkFactory.QOS);
62-
this.retryTimeoutMs = pluginConfig.get(MqttSinkFactory.RETRY_TIMEOUT);
60+
this.topic = pluginConfig.get(MqttSinkOptions.TOPIC);
61+
this.qos = pluginConfig.get(MqttSinkOptions.QOS);
62+
this.retryTimeoutMs = pluginConfig.get(MqttSinkOptions.RETRY_TIMEOUT);
6363
this.serializationSchema = createSerializationSchema(rowType, pluginConfig);
6464

6565
// Each subtask appends its index to guarantee a unique MQTT client ID,
@@ -70,7 +70,7 @@ public MqttSinkWriter(
7070
// MemoryPersistence avoids file-system I/O; ideal for containerized deployments.
7171
this.mqttClient =
7272
new MqttClient(
73-
pluginConfig.get(MqttSinkFactory.URL),
73+
pluginConfig.get(MqttSinkOptions.URL),
7474
clientId,
7575
new MemoryPersistence());
7676
this.mqttClient.setCallback(this);
@@ -80,7 +80,7 @@ public MqttSinkWriter(
8080
log.info(
8181
"MQTT sink writer [{}] connected to {}",
8282
clientId,
83-
pluginConfig.get(MqttSinkFactory.URL));
83+
pluginConfig.get(MqttSinkOptions.URL));
8484
} catch (MqttException e) {
8585
throw new RuntimeException("Failed to connect MQTT client [" + clientId + "]", e);
8686
}
@@ -166,13 +166,13 @@ private static MqttConnectOptions buildConnectOptions(ReadonlyConfig config) {
166166
MqttConnectOptions options = new MqttConnectOptions();
167167
options.setAutomaticReconnect(true);
168168
options.setCleanSession(true);
169-
options.setConnectionTimeout(config.get(MqttSinkFactory.CONNECTION_TIMEOUT));
169+
options.setConnectionTimeout(config.get(MqttSinkOptions.CONNECTION_TIMEOUT));
170170

171-
String username = config.get(MqttSinkFactory.USERNAME);
171+
String username = config.get(MqttSinkOptions.USERNAME);
172172
if (username != null && !username.isEmpty()) {
173173
options.setUserName(username);
174174
}
175-
String password = config.get(MqttSinkFactory.PASSWORD);
175+
String password = config.get(MqttSinkOptions.PASSWORD);
176176
if (password != null && !password.isEmpty()) {
177177
options.setPassword(password.toCharArray());
178178
}
@@ -181,7 +181,7 @@ private static MqttConnectOptions buildConnectOptions(ReadonlyConfig config) {
181181

182182
private static SerializationSchema createSerializationSchema(
183183
SeaTunnelRowType rowType, ReadonlyConfig config) {
184-
String format = config.get(MqttSinkFactory.FORMAT);
184+
String format = config.get(MqttSinkOptions.FORMAT);
185185
switch (format.toLowerCase()) {
186186
case "json":
187187
return new JsonSerializationSchema(rowType);

‎seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mqtt-e2e/src/test/resources/mosquitto.conf‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,19 @@
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+
117
# Minimal Mosquitto configuration for E2E testing.
218
# Allow anonymous connections on port 1883.
319
listener 1883

0 commit comments

Comments
 (0)