# Dynamically sink the data to multiple kafka topics using mqtt source connector

**URL:** https://ask.lenses.io/t/dynamically-sink-the-data-to-multiple-kafka-topics-using-mqtt-source-connector/348
**Category:** Apache Kafka Connectors
**Tags:** kafka-connect
**Created:** [April 3, 2025, 11:54am UTC](https://ask.lenses.io/t/dynamically-sink-the-data-to-multiple-kafka-topics-using-mqtt-source-connector/348 "2025-04-03T11:54:59Z")
**Posts on this page:** 6
**Page:** 1

<div class="post-metadata">

### Author: ![abhishekgoyal](https://avatars.discourse-cdn.com/v4/letter/a/b3f665/32.png) [@abhishekgoyal](https://ask.lenses.io/u/abhishekgoyal)
#### Post date: [April 3, 2025, 11:54am UTC](https://ask.lenses.io/t/dynamically-sink-the-data-to-multiple-kafka-topics-using-mqtt-source-connector/348/1 "2025-04-03T11:54:59Z")

</div>

I am new to this tool. I have a requirement in which I need to dynamically sink the data to multiple kafka topics using mqtt source connector. For that, I have used the kafka CRD like

```auto
spec:
  class: io.lenses.streamreactor.connect.mqtt.source.MqttSourceConnector
  config:
    tasks.max: 1
    connect.mqtt.clean: "true"
    connect.mqtt.hosts: "tcp://rabbitmq-cluster:1883"
    connect.mqtt.converter.throw.on.error: "true"
    connect.mqtt.log.message: "true"
    connect.mqtt.client.id: "lenses-client-connector"
    connect.mqtt.timeout: "1"
    connect.mqtt.keep.alive: "2"
    connect.mqtt.username: "<rabbitmq-client-id>"
    connect.mqtt.password: "<rabbitmq-client-secret>"
    connect.mqtt.service.quality: "1"
    connect.key.converter: "org.apache.kafka.connect.storage.StringConverter"
    connect.value.converter: "org.apache.kafka.connect.converters.StringConverter"
    connect.mqtt.kcql: "INSERT INTO <kafka-topic> SELECT * FROM `<queue-name>` WITHCONVERTER=`io.lenses.streamreactor.connect.converters.source.JsonSimpleConverter`"

```

I have check on the ChatGPT it suggests that we can use `"INSERT INTO kafka_${1} SELECT * FROM mqtt/topic/+/device/+/+"` but i further check with the lenses stream reactor documentation, I have found no way like this.

Can someone suggest how can I achieve my usecase?

Thanks

---

<div class="post-metadata">

### Author: ![Andrew\_Stevenson](https://dub1.discourse-cdn.com/flex017/user_avatar/ask.lenses.io/andrew_stevenson/32/80_2.png) [@Andrew\_Stevenson](https://ask.lenses.io/u/Andrew_Stevenson)
#### Post date: [April 10, 2025, 10:28am UTC](https://ask.lenses.io/t/dynamically-sink-the-data-to-multiple-kafka-topics-using-mqtt-source-connector/348/2 "2025-04-10T10:28:07Z")

</div>

You use the dynamic target functionality. This with route to a topic based on a field value. You can use SMTs to add or modify fields.

[MQTT | Lenses Docs](https://docs.lenses.io/latest/connectors/kafka-connectors/sinks/mqtt#dynamic-targets)

---

<div class="post-metadata">

### Author: ![abhishekgoyal](https://avatars.discourse-cdn.com/v4/letter/a/b3f665/32.png) [@abhishekgoyal](https://ask.lenses.io/u/abhishekgoyal)
#### Post date: [April 10, 2025, 11:17am UTC](https://ask.lenses.io/t/dynamically-sink-the-data-to-multiple-kafka-topics-using-mqtt-source-connector/348/3 "2025-04-10T11:17:17Z")

</div>

Thank you for your response. But, in my use case, we don’t receive the kafka topic name in the field, but we’ll receive it in the mqtt topic name itself. So, are there any way to do this. Also, if you have some documents regarding SMTs because I have tried to find them but got no documentations.  
Thanks.

---

<div class="post-metadata">

### Author: ![Andrew\_Stevenson](https://dub1.discourse-cdn.com/flex017/user_avatar/ask.lenses.io/andrew_stevenson/32/80_2.png) [@Andrew\_Stevenson](https://ask.lenses.io/u/Andrew_Stevenson)
#### Post date: [April 10, 2025, 12:07pm UTC](https://ask.lenses.io/t/dynamically-sink-the-data-to-multiple-kafka-topics-using-mqtt-source-connector/348/4 "2025-04-10T12:07:41Z")

</div>

Ah, sorry I thought it was the sink.

For the source you can use the wildcard subscription and set the KCQL target to `$`. I will update the docs but the test case his here. [stream-reactor/kafka-connect-mqtt/src/it/scala/io/lenses/streamreactor/connect/mqtt/source/MqttManagerTest.scala at master · lensesio/stream-reactor](https://github.com/lensesio/stream-reactor/blob/master/kafka-connect-mqtt/src/it/scala/io/lenses/streamreactor/connect/mqtt/source/MqttManagerTest.scala#L93)

---

<div class="post-metadata">

### Author: ![abhishekgoyal](https://avatars.discourse-cdn.com/v4/letter/a/b3f665/32.png) [@abhishekgoyal](https://ask.lenses.io/u/abhishekgoyal)
#### Post date: [April 11, 2025, 3:12am UTC](https://ask.lenses.io/t/dynamically-sink-the-data-to-multiple-kafka-topics-using-mqtt-source-connector/348/5 "2025-04-11T03:12:20Z")

</div>

Thank you for your response, It will dynamically route to a kafka topic same as mqtt topic. But I need to extract only one string from my mqtt pattern.  
For example:  
MQTT Topic Pattern - mqtt/topic/+/device/+/+  
MQTT Topic Example 1 - mqtt/topic/A/device/B/C  
MQTT Topic Example 2 - mqtt/topic/X/device/Y/Z

The kafka topic needs to create like - kafka\_A, kafka\_X

Can you help me out with this?

---

<div class="post-metadata">

### Author: ![Andrew\_Stevenson](https://dub1.discourse-cdn.com/flex017/user_avatar/ask.lenses.io/andrew_stevenson/32/80_2.png) [@Andrew\_Stevenson](https://ask.lenses.io/u/Andrew_Stevenson)
#### Post date: [April 23, 2025, 11:28am UTC](https://ask.lenses.io/t/dynamically-sink-the-data-to-multiple-kafka-topics-using-mqtt-source-connector/348/6 "2025-04-23T11:28:37Z")

</div>

Have you tried an SMT, for source connectors they sit between the connector and kafka and can manipulate the records including the topic name.

[Single Message Transforms | Redpanda Cloud](https://docs.redpanda.com/redpanda-cloud/develop/managed-connectors/transforms/#regexrouter)