You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "Dream95 (via GitHub)" <gi...@apache.org> on 2026/06/24 06:57:36 UTC
[PR] [feat][mqtt] Support dynamic MQTT topic via topicProperty config [pulsar-connectors]
Dream95 opened a new pull request, #34:
URL: https://github.com/apache/pulsar-connectors/pull/34
**Motivation**
In MQTT IoT setups, messages are often routed per device to topics like /data/{deviceId} or sensor/{deviceId}/telemetry. Each device may need its own MQTT topic instead of a single shared one.
MqttSink only supports a fixed topic in configuration today, so it cannot choose the destination topic per message from device identity.
**Solution**
Add an optional topicProperty setting to MqttSink. It names the Pulsar message property that holds the target MQTT topic.
When topicProperty is set, the sink reads that key from record.getProperties() and uses the value as the publish topic. If topicProperty is unset, or the property is missing or blank on a message, the sink uses the configured topic. Existing configs that only set the topic behave the same as before.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][mqtt] Support dynamic MQTT topic via topicProperty config [pulsar-connectors]
Posted by "david-streamlio (via GitHub)" <gi...@apache.org>.
david-streamlio merged PR #34:
URL: https://github.com/apache/pulsar-connectors/pull/34
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][mqtt] Support dynamic MQTT topic via topicProperty config [pulsar-connectors]
Posted by "david-streamlio (via GitHub)" <gi...@apache.org>.
david-streamlio commented on PR #34:
URL: https://github.com/apache/pulsar-connectors/pull/34#issuecomment-4936394520
Context that may be useful for this PR: as of #74, CI now runs the `integrationTest` source set, so `mqtt`'s existing `MqttSinkIntegrationTest` (a real `eclipse-mosquitto:2` container, added alongside the connector in #22) executes on every pull request. Before #74 it had never run on a PR.
That test covers only the static `mqttSinkConfig.getTopic()` path. The dynamic-topic behavior this PR adds via `topicProperty` would be a natural second case there — a record carrying the property should land on the derived topic, and one without it should fall back to the configured topic. Noted while closing #55, which turned out to be about the CI gap rather than missing coverage.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][mqtt] Support dynamic MQTT topic via topicProperty config [pulsar-connectors]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #34:
URL: https://github.com/apache/pulsar-connectors/pull/34#discussion_r3553863935
##########
mqtt/src/main/java/org/apache/pulsar/io/mqtt/MqttSink.java:
##########
@@ -106,16 +107,25 @@ public void write(Record<byte[]> record) {
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;
+ }
+ String dynamicTopic = record.getProperties().get(topicProperty);
+ return StringUtils.isNotBlank(dynamicTopic) ? dynamicTopic : defaultTopic;
+ }
Review Comment:
resolveTopic() assumes record.getProperties() is non-null. Other connectors in this repo defensively handle null properties (e.g., kinesis Utils.serializeRecordToJson checks for null), so a Record implementation that returns null here will throw NPE before the write() try/catch runs, skipping record.fail() and dropping the intended fallback-to-default behavior.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org