You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by da...@apache.org on 2026/07/14 15:56:06 UTC
(pulsar-connectors) branch master updated: [feat][mqtt] Support dynamic MQTT topic via topicProperty config (#34)
This is an automated email from the ASF dual-hosted git repository.
david-streamlio pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar-connectors.git
The following commit(s) were added to refs/heads/master by this push:
new 5d5847ae [feat][mqtt] Support dynamic MQTT topic via topicProperty config (#34)
5d5847ae is described below
commit 5d5847aed78b64990648a4a8253f90e73ba81a90
Author: Dream95 <86...@qq.com>
AuthorDate: Tue Jul 14 23:56:00 2026 +0800
[feat][mqtt] Support dynamic MQTT topic via topicProperty config (#34)
* [feat][mqtt] Support dynamic MQTT topic via topicProperty config
Signed-off-by: Dream95 <zh...@163.com>
* [fix][mqtt] Fall back to default topic when record properties is null
Signed-off-by: Dream95 <zh...@163.com>
* [improve][test] Add MQTT sink e2e coverage for dynamic topicProperty
Signed-off-by: Dream95 <zh...@163.com>
* [fix][test] Add timeOut to MQTT dynamic topicProperty e2e test
Signed-off-by: Dream95 <zh...@163.com>
---------
Signed-off-by: Dream95 <zh...@163.com>
Co-authored-by: David Kjerrumgaard <35...@users.noreply.github.com>
---
.../pulsar/io/mqtt/MqttSinkIntegrationTest.java | 132 ++++++++++++++++-----
.../java/org/apache/pulsar/io/mqtt/MqttSink.java | 22 +++-
.../org/apache/pulsar/io/mqtt/MqttSinkConfig.java | 8 +-
.../apache/pulsar/io/mqtt/MqttSinkConfigTest.java | 10 ++
.../org/apache/pulsar/io/mqtt/MqttSinkTest.java | 65 +++++++++-
5 files changed, 202 insertions(+), 35 deletions(-)
diff --git a/mqtt/src/integrationTest/java/org/apache/pulsar/io/mqtt/MqttSinkIntegrationTest.java b/mqtt/src/integrationTest/java/org/apache/pulsar/io/mqtt/MqttSinkIntegrationTest.java
index 101d3776..49e7294a 100644
--- a/mqtt/src/integrationTest/java/org/apache/pulsar/io/mqtt/MqttSinkIntegrationTest.java
+++ b/mqtt/src/integrationTest/java/org/apache/pulsar/io/mqtt/MqttSinkIntegrationTest.java
@@ -26,6 +26,7 @@ import com.hivemq.client.mqtt.MqttClient;
import com.hivemq.client.mqtt.mqtt5.Mqtt5AsyncClient;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
+import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
@@ -75,55 +76,116 @@ public class MqttSinkIntegrationTest {
CountDownLatch ackLatch = new CountDownLatch(3);
AtomicBoolean failCalled = new AtomicBoolean(false);
+ Mqtt5AsyncClient subscriber = connectSubscriber("mqtt-sink-e2e-subscriber");
+ try {
+ subscribe(subscriber, TEST_TOPIC, receivedPayloads);
+
+ Map<String, Object> config = baseSinkConfig(TEST_TOPIC);
+ config.put("clientId", "mqtt-sink-e2e-publisher");
+
+ try (MqttSink sink = new MqttSink()) {
+ sink.open(config, mock(SinkContext.class));
+
+ for (int i = 0; i < 3; i++) {
+ sink.write(new TestRecord(("msg-" + i).getBytes(StandardCharsets.UTF_8), ackLatch, failCalled));
+ }
+
+ assertTrue(ackLatch.await(10, TimeUnit.SECONDS), "Timed out waiting for record.ack()");
+ assertFalse(failCalled.get(), "record.fail() should not be called on successful publish");
+
+ assertEquals(receivedPayloads.poll(10, TimeUnit.SECONDS), "msg-0");
+ assertEquals(receivedPayloads.poll(10, TimeUnit.SECONDS), "msg-1");
+ assertEquals(receivedPayloads.poll(10, TimeUnit.SECONDS), "msg-2");
+ }
+ } finally {
+ disconnectQuietly(subscriber);
+ }
+ }
+
+ @Test(timeOut = 120_000)
+ public void testWriteE2EWithDynamicTopicProperty() throws Exception {
+ String defaultTopic = "pulsar/mqtt/e2e/default";
+ String dynamicTopic = "pulsar/mqtt/e2e/device-1";
+ BlockingQueue<String> defaultPayloads = new LinkedBlockingQueue<>();
+ BlockingQueue<String> dynamicPayloads = new LinkedBlockingQueue<>();
+ CountDownLatch ackLatch = new CountDownLatch(2);
+ AtomicBoolean failCalled = new AtomicBoolean(false);
+
+ Mqtt5AsyncClient subscriber = connectSubscriber("mqtt-sink-e2e-dynamic-subscriber");
+ try {
+ subscribe(subscriber, defaultTopic, defaultPayloads);
+ subscribe(subscriber, dynamicTopic, dynamicPayloads);
+
+ Map<String, Object> config = baseSinkConfig(defaultTopic);
+ config.put("topicProperty", "mqttTopic");
+ config.put("clientId", "mqtt-sink-e2e-dynamic-publisher");
+
+ try (MqttSink sink = new MqttSink()) {
+ sink.open(config, mock(SinkContext.class));
+
+ sink.write(new TestRecord(
+ "dynamic-msg".getBytes(StandardCharsets.UTF_8),
+ ackLatch,
+ failCalled,
+ Map.of("mqttTopic", dynamicTopic)));
+ sink.write(new TestRecord(
+ "default-msg".getBytes(StandardCharsets.UTF_8),
+ ackLatch,
+ failCalled));
+
+ assertTrue(ackLatch.await(10, TimeUnit.SECONDS), "Timed out waiting for record.ack()");
+ assertFalse(failCalled.get(), "record.fail() should not be called on successful publish");
+
+ assertEquals(dynamicPayloads.poll(10, TimeUnit.SECONDS), "dynamic-msg");
+ assertEquals(defaultPayloads.poll(10, TimeUnit.SECONDS), "default-msg");
+ }
+ } finally {
+ disconnectQuietly(subscriber);
+ }
+ }
+
+ private Mqtt5AsyncClient connectSubscriber(String clientId) throws Exception {
Mqtt5AsyncClient subscriber = MqttClient.builder()
.useMqttVersion5()
.serverHost(mqttContainer.getHost())
.serverPort(mqttContainer.getMappedPort(MQTT_PORT))
- .identifier("mqtt-sink-e2e-subscriber")
+ .identifier(clientId)
.buildAsync();
-
subscriber.connectWith()
.cleanStart(true)
.send()
.get(10, TimeUnit.SECONDS);
+ return subscriber;
+ }
+
+ private void subscribe(Mqtt5AsyncClient subscriber, String topic,
+ BlockingQueue<String> payloads) throws Exception {
subscriber.subscribeWith()
- .topicFilter(TEST_TOPIC)
- .callback(publish -> receivedPayloads.add(
+ .topicFilter(topic)
+ .callback(publish -> payloads.add(
new String(publish.getPayloadAsBytes(), StandardCharsets.UTF_8)))
.send()
.get(10, TimeUnit.SECONDS);
+ }
+ private Map<String, Object> baseSinkConfig(String topic) {
Map<String, Object> config = new HashMap<>();
config.put("serverHost", mqttContainer.getHost());
config.put("serverPort", mqttContainer.getMappedPort(MQTT_PORT));
- config.put("topic", TEST_TOPIC);
+ config.put("topic", topic);
config.put("qos", 1);
config.put("connectionTimeoutMs", 10000);
- config.put("clientId", "mqtt-sink-e2e-publisher");
-
- SinkContext sinkContext = mock(SinkContext.class);
- try (MqttSink sink = new MqttSink()) {
- sink.open(config, sinkContext);
-
- for (int i = 0; i < 3; i++) {
- sink.write(new TestRecord(("msg-" + i).getBytes(StandardCharsets.UTF_8), ackLatch, failCalled));
- }
-
- assertTrue(ackLatch.await(10, TimeUnit.SECONDS), "Timed out waiting for record.ack()");
- assertFalse(failCalled.get(), "record.fail() should not be called on successful publish");
+ return config;
+ }
- assertEquals(receivedPayloads.poll(10, TimeUnit.SECONDS), "msg-0");
- assertEquals(receivedPayloads.poll(10, TimeUnit.SECONDS), "msg-1");
- assertEquals(receivedPayloads.poll(10, TimeUnit.SECONDS), "msg-2");
- } finally {
- try {
- subscriber.disconnectWith()
- .sessionExpiryInterval(0)
- .send()
- .get(10, TimeUnit.SECONDS);
- } catch (Exception e) {
- log.warn("Failed to disconnect MQTT subscriber in test cleanup", e);
- }
+ private void disconnectQuietly(Mqtt5AsyncClient subscriber) {
+ try {
+ subscriber.disconnectWith()
+ .sessionExpiryInterval(0)
+ .send()
+ .get(10, TimeUnit.SECONDS);
+ } catch (Exception e) {
+ log.warn("Failed to disconnect MQTT subscriber in test cleanup", e);
}
}
@@ -131,11 +193,18 @@ public class MqttSinkIntegrationTest {
private final byte[] value;
private final CountDownLatch ackLatch;
private final AtomicBoolean failCalled;
+ private final Map<String, String> properties;
private TestRecord(byte[] value, CountDownLatch ackLatch, AtomicBoolean failCalled) {
+ this(value, ackLatch, failCalled, Collections.emptyMap());
+ }
+
+ private TestRecord(byte[] value, CountDownLatch ackLatch, AtomicBoolean failCalled,
+ Map<String, String> properties) {
this.value = value;
this.ackLatch = ackLatch;
this.failCalled = failCalled;
+ this.properties = properties;
}
@Override
@@ -143,6 +212,11 @@ public class MqttSinkIntegrationTest {
return value;
}
+ @Override
+ public Map<String, String> getProperties() {
+ return properties;
+ }
+
@Override
public void ack() {
ackLatch.countDown();
diff --git a/mqtt/src/main/java/org/apache/pulsar/io/mqtt/MqttSink.java b/mqtt/src/main/java/org/apache/pulsar/io/mqtt/MqttSink.java
index cf7eee40..30d47e93 100644
--- a/mqtt/src/main/java/org/apache/pulsar/io/mqtt/MqttSink.java
+++ b/mqtt/src/main/java/org/apache/pulsar/io/mqtt/MqttSink.java
@@ -94,10 +94,11 @@ public class MqttSink implements Sink<byte[]> {
@Override
public void write(Record<byte[]> record) {
+ String topic = resolveTopic(record);
try {
byte[] payload = record.getValue() == null ? new byte[0] : record.getValue();
mqttClient.publishWith()
- .topic(mqttSinkConfig.getTopic())
+ .topic(topic)
.qos(mqttQos)
.payload(payload)
.send()
@@ -106,16 +107,29 @@ public class MqttSink implements Sink<byte[]> {
record.ack();
} else {
record.fail();
- log.warn("Failed to publish message to MQTT topic {}",
- mqttSinkConfig.getTopic(), throwable);
+ log.warn("Failed to publish message to MQTT topic {}", topic, throwable);
}
});
} catch (Exception e) {
record.fail();
- log.warn("Failed to schedule MQTT publish for topic {}", mqttSinkConfig.getTopic(), e);
+ log.warn("Failed to schedule MQTT publish for topic {}", topic, e);
}
}
+ String resolveTopic(Record<byte[]> record) {
+ String defaultTopic = mqttSinkConfig.getTopic();
+ String topicProperty = mqttSinkConfig.getTopicProperty();
+ if (StringUtils.isBlank(topicProperty)) {
+ return defaultTopic;
+ }
+ Map<String, String> properties = record.getProperties();
+ if (properties == null) {
+ return defaultTopic;
+ }
+ String dynamicTopic = properties.get(topicProperty);
+ return StringUtils.isNotBlank(dynamicTopic) ? dynamicTopic : defaultTopic;
+ }
+
@Override
public void close() {
if (mqttClient == null) {
diff --git a/mqtt/src/main/java/org/apache/pulsar/io/mqtt/MqttSinkConfig.java b/mqtt/src/main/java/org/apache/pulsar/io/mqtt/MqttSinkConfig.java
index 646462d0..acbd0245 100644
--- a/mqtt/src/main/java/org/apache/pulsar/io/mqtt/MqttSinkConfig.java
+++ b/mqtt/src/main/java/org/apache/pulsar/io/mqtt/MqttSinkConfig.java
@@ -52,9 +52,15 @@ public class MqttSinkConfig implements Serializable {
@FieldDoc(
required = true,
defaultValue = "",
- help = "The MQTT topic to publish messages to.")
+ help = "The default MQTT topic to publish messages to when no dynamic topic is available.")
private String topic;
+ @FieldDoc(
+ defaultValue = "",
+ help = "The Pulsar message property key used to resolve the MQTT topic dynamically. "
+ + "When the property is missing or blank, the configured topic is used as the default.")
+ private String topicProperty;
+
@FieldDoc(
defaultValue = "",
help = "MQTT client id used for the broker connection.")
diff --git a/mqtt/src/test/java/org/apache/pulsar/io/mqtt/MqttSinkConfigTest.java b/mqtt/src/test/java/org/apache/pulsar/io/mqtt/MqttSinkConfigTest.java
index 0707f0a7..26615fed 100644
--- a/mqtt/src/test/java/org/apache/pulsar/io/mqtt/MqttSinkConfigTest.java
+++ b/mqtt/src/test/java/org/apache/pulsar/io/mqtt/MqttSinkConfigTest.java
@@ -67,6 +67,16 @@ public class MqttSinkConfigTest {
assertEquals(config.getQos(), 1);
}
+ @Test
+ public void loadTopicPropertyFromMapTest() throws IOException {
+ Map<String, Object> map = baseConfigMap();
+ map.put("topicProperty", "mqttTopic");
+ SinkContext sinkContext = Mockito.mock(SinkContext.class);
+ MqttSinkConfig config = MqttSinkConfig.load(map, sinkContext);
+
+ assertEquals(config.getTopicProperty(), "mqttTopic");
+ }
+
@Test
public void loadFromMapCredentialsFromSecretTest() throws IOException {
Map<String, Object> map = baseConfigMap();
diff --git a/mqtt/src/test/java/org/apache/pulsar/io/mqtt/MqttSinkTest.java b/mqtt/src/test/java/org/apache/pulsar/io/mqtt/MqttSinkTest.java
index 326d5a97..00f3d90b 100644
--- a/mqtt/src/test/java/org/apache/pulsar/io/mqtt/MqttSinkTest.java
+++ b/mqtt/src/test/java/org/apache/pulsar/io/mqtt/MqttSinkTest.java
@@ -22,10 +22,12 @@ import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertTrue;
import com.hivemq.client.mqtt.mqtt5.Mqtt5AsyncClient;
import com.hivemq.client.mqtt.mqtt5.message.connect.Mqtt5ConnectBuilder;
import com.hivemq.client.mqtt.mqtt5.message.connect.connack.Mqtt5ConnAck;
+import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
@@ -66,7 +68,56 @@ public class MqttSinkTest {
new MqttSink().close();
}
+ @Test
+ public void resolveTopicShouldUseDynamicTopicFromRecordProperty() {
+ Map<String, Object> config = baseConfigMap();
+ config.put("topicProperty", "mqttTopic");
+ MqttSink sink = newSinkWithOpenedClient(mock(Mqtt5AsyncClient.class), config);
+
+ TestRecord record = new TestRecord(
+ "x".getBytes(),
+ new CountDownLatch(1),
+ new AtomicBoolean(false),
+ Map.of("mqttTopic", "dynamic/topic"));
+
+ assertEquals(sink.resolveTopic(record), "dynamic/topic");
+ }
+
+ @Test
+ public void resolveTopicShouldFallBackToDefaultTopicWhenPropertyMissing() {
+ Map<String, Object> config = baseConfigMap();
+ config.put("topicProperty", "mqttTopic");
+ MqttSink sink = newSinkWithOpenedClient(mock(Mqtt5AsyncClient.class), config);
+
+ TestRecord record = new TestRecord(
+ "x".getBytes(),
+ new CountDownLatch(1),
+ new AtomicBoolean(false),
+ Collections.emptyMap());
+
+ assertEquals(sink.resolveTopic(record), "test/topic");
+ }
+
+ @Test
+ public void resolveTopicShouldFallBackToDefaultTopicWhenPropertiesNull() {
+ Map<String, Object> config = baseConfigMap();
+ config.put("topicProperty", "mqttTopic");
+ MqttSink sink = newSinkWithOpenedClient(mock(Mqtt5AsyncClient.class), config);
+
+ TestRecord record = new TestRecord(
+ "x".getBytes(),
+ new CountDownLatch(1),
+ new AtomicBoolean(false),
+ null);
+
+ assertEquals(sink.resolveTopic(record), "test/topic");
+ }
+
private MqttSink newSinkWithOpenedClient(Mqtt5AsyncClient mqttClient) {
+ return newSinkWithOpenedClient(mqttClient, baseConfigMap());
+ }
+
+ private MqttSink newSinkWithOpenedClient(Mqtt5AsyncClient mqttClient, Map<String, Object> config) {
try {
@SuppressWarnings("unchecked")
Mqtt5ConnectBuilder.Send<CompletableFuture<Mqtt5ConnAck>> connectBuilder =
@@ -77,7 +128,7 @@ public class MqttSinkTest {
MqttSink sink = Mockito.spy(new MqttSink());
doReturn(mqttClient).when(sink).buildClient(any());
- sink.open(baseConfigMap(), mock(SinkContext.class));
+ sink.open(config, mock(SinkContext.class));
return sink;
} catch (Exception e) {
throw new AssertionError("Failed to initialize MqttSink test fixture", e);
@@ -100,11 +151,18 @@ public class MqttSinkTest {
private final byte[] value;
private final CountDownLatch ackLatch;
private final AtomicBoolean failCalled;
+ private final Map<String, String> properties;
private TestRecord(byte[] value, CountDownLatch ackLatch, AtomicBoolean failCalled) {
+ this(value, ackLatch, failCalled, Collections.emptyMap());
+ }
+
+ private TestRecord(byte[] value, CountDownLatch ackLatch, AtomicBoolean failCalled,
+ Map<String, String> properties) {
this.value = value;
this.ackLatch = ackLatch;
this.failCalled = failCalled;
+ this.properties = properties;
}
@Override
@@ -112,6 +170,11 @@ public class MqttSinkTest {
return value;
}
+ @Override
+ public Map<String, String> getProperties() {
+ return properties;
+ }
+
@Override
public void ack() {
ackLatch.countDown();