# MQTT source connector byte\[\] array transform to JSON

**URL:** <https://forum.confluent.io/t/mqtt-source-connector-byte-array-transform-to-json/37690>\
**Category:** Kafka Connect\
**Created:** [23 April 2025 09:13 UTC](https://forum.confluent.io/t/mqtt-source-connector-byte-array-transform-to-json/37690 "2025-04-23T09:13:03Z")\
**Posts on this page:** 7\
**Page:** 1

<div class="post-metadata">

**Author:** ![Metalmania97](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/metalmania97/32/4787_2.png) [@Metalmania97](https://forum.confluent.io/u/Metalmania97)\
**Post date:** [23 April 2025 09:13 UTC](https://forum.confluent.io/t/mqtt-source-connector-byte-array-transform-to-json/37690/1 "2025-04-23T09:13:03Z")

</div>

Hello. I have an MQTT producer that sends the message payload as a byte array to the MQTT source connector. Is there a way for the connector to transform the byte array message to actual json so that the messages follow the topic schema?  
I have this configuration for the connector:  
{  
“name”: “MqttSourceConnectorTemp”,  
“connector.class”: “io.confluent.connect.mqtt.MqttSourceConnector”,  
“mqtt.qos”: “0”,  
“confluent.topic.bootstrap.servers”: “kafka:9092”,  
“tasks.max”: “1”,  
“mqtt.clean.session.enabled”: “true”,  
“transaction.boundary”: “poll”,  
“mqtt.server.uri”: “tcp://mqtt-broker:1883”,  
“header.converter”: “org.apache.kafka.connect.storage.StringConverter”,  
“key.converter.schemas.enable”: “false”,  
“exactly.once.support”: “requested”,  
“confluent.topic.replication.factor”: “1”,  
“value.converter.schemas.enable”: “false”,  
“mqtt.topics”: “mqtt/temp/#”,  
“kafka.topic”: “temp”,  
“value.converter”: “org.apache.kafka.connect.converters.ByteArrayConverter”,  
“key.converter”: “org.apache.kafka.connect.storage.StringConverter”  
}

---

<div class="post-metadata">

**Author:** ![dtroiano](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/dtroiano/32/1961_2.png) [@dtroiano](https://forum.confluent.io/u/dtroiano)\
**Post date:** [23 April 2025 16:43 UTC](https://forum.confluent.io/t/mqtt-source-connector-byte-array-transform-to-json/37690/2 "2025-04-23T16:43:34Z")

</div>

> [@Metalmania97](#):
>
> Is there a way for the connector to transform the byte array message to actual json so that the messages follow the topic schema?

Can you explain the serialization process? How does the byte array get created?

On the Kafka side, are you using Schema Registry and [JSON Schema](https://docs.confluent.io/platform/current/schema-registry/fundamentals/serdes-develop/serdes-json.html#)-formatted values? Or just JSON strings?

---

<div class="post-metadata">

**Author:** ![Metalmania97](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/metalmania97/32/4787_2.png) [@Metalmania97](https://forum.confluent.io/u/Metalmania97)\
**Post date:** [29 April 2025 15:10 UTC](https://forum.confluent.io/t/mqtt-source-connector-byte-array-transform-to-json/37690/3 "2025-04-29T15:10:28Z")

</div>

I have a java producer that creates a json message and sends it to the MQTT broker and then through the MQTT source connector. The MqttMessage method of the Eclipse Paho Client library requires a byte payload so i set “value.converter”: “org.apache.kafka.connect.converters.ByteArrayConverter”. What can I do to convert the byte array to proper json? I use ksqldb and kafka streams later in the pipeline that use json\_sr as the value format but i want the input topic to be proper json as well to be able to write the messages to a database using the JDBC sink connector. As it is now I can’t use the JDBC sink connector with a byte array to map the json fields to the table columns

Random randomOffset = new Random();  
Double randomTemperature = randomOffset.nextDouble(86,88);  
JsonObject json = new JsonObject();  
json.addProperty(“message\_id”, UUID.randomUUID().toString());  
json.addProperty(“temperature”, randomTemperature);  
json.addProperty(“timestamp”, ZonedDateTime.now(ZoneOffset.UTC).format(formatter).toString());

// Convert JSON to String and set as payload  
content = json.toString();  
MqttMessage message = new MqttMessage(content.getBytes(StandardCharsets.UTF\_8));  
message.setQos(0);  
client.publish(topic, message);

---

<div class="post-metadata">

**Author:** ![dtroiano](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/dtroiano/32/1961_2.png) [@dtroiano](https://forum.confluent.io/u/dtroiano)\
**Post date:** [29 April 2025 18:09 UTC](https://forum.confluent.io/t/mqtt-source-connector-byte-array-transform-to-json/37690/4 "2025-04-29T18:09:21Z")

</div>

A few ideas come to mind:

1. Check out [this](https://jcustenborder.github.io/kafka-connect-documentation/projects/kafka-connect-transform-common/transformations/examples/BytesToString.struct.html) `BytesToString` SMT ([GitHub](https://github.com/jcustenborder/kafka-connect-transform-common/)) that can be added to the connector config to get you back to the string in your MQTT client code. You’d use `org.apache.kafka.connect.storage.StringConverter` instead. This wouldn’t give you JSON Schema but would get you to something closer / potentially easier to play with.
2. Generate the proper JSON Schema-formatted message on the MQTT client side. [Here](https://docs.confluent.io/platform/current/schema-registry/fundamentals/serdes-develop/index.html#wire-format) is the wire format doc. So you would write a zero byte, then 4 bytes for schema ID, then the result of [KafkaJsonSchemaSerializer.serialize()](https://github.com/confluentinc/schema-registry/blob/184235a60a9bd4e22d9a34f3b8fdf480abda113c/json-schema-serializer/src/main/java/io/confluent/kafka/serializers/json/KafkaJsonSchemaSerializer.java#L32). Keep using `ByteArrayConverter` in the connector config.
3. Land the data as is in a byte array and then write a stream processing app to convert. E.g., [here](https://github.com/confluentinc/tutorials/blob/c769f7763f42f46732292bfa1ea92a532ecde50a/serialization/kstreams/src/main/java/io/confluent/developer/AvroToProtobuf.java) is a Kafka Streams app to convert from Avro to Protobuf that could be a starting point for converting from UTF-8 encoded string to JSON Schema format. Obviously adds overhead given the topic-to-topic transformation compared to option 2

I would give option 2 a try first. It feels a bit hacky and I could be missing something that would prevent it from working, but the efficiency plus getting into a Schema Regsitry format from the get go makes it worthwhile IMO (assuming it works 🙂 )

---

<div class="post-metadata">

**Author:** ![Metalmania97](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/metalmania97/32/4787_2.png) [@Metalmania97](https://forum.confluent.io/u/Metalmania97)\
**Post date:** [29 April 2025 21:35 UTC](https://forum.confluent.io/t/mqtt-source-connector-byte-array-transform-to-json/37690/5 "2025-04-29T21:35:00Z")

</div>

I try to simulate a real IoT scenario so I think idea 1 is the best solution!? Would I be able to use the JDBC sink connector and dump the json string messages in a single/multiple column table? If I set a schema to the topic that has json string messages would that be sufficient?

---

<div class="post-metadata">

**Author:** ![dtroiano](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/dtroiano/32/1961_2.png) [@dtroiano](https://forum.confluent.io/u/dtroiano)\
**Post date:** [30 April 2025 15:25 UTC](https://forum.confluent.io/t/mqtt-source-connector-byte-array-transform-to-json/37690/6 "2025-04-30T15:25:56Z")

</div>

Instead of #1 there is a much better `FromJson` SMT [here](https://jcustenborder.github.io/kafka-connect-documentation/projects/kafka-connect-json-schema/transformations/FromJson.html) ([GitHub](https://github.com/jcustenborder/kafka-connect-json-schema)) for what you want to do. The Confluent Platform `cp-demo` example also uses it [here](https://github.com/confluentinc/cp-demo/blob/685b58298cf11841b2097c43b3c8c06d56e3d257/scripts/connectors/submit_wikipedia_sse_config.sh).

You can use it on the source or the sink.

To use on the source, since the record flow is SMT → converter, you would use this SMT and then one of the Schema Registry converters as the CP demo example does.

To use on the sink, since the record flow is converter → SMT, assuming you used `ByteArrayConverter` in the `MqttSourceConnector`, you would also use `ByteArrayConverter` on the sink. [Here](https://github.com/confluentinc/intesa-usecases/blob/bbbefd23b656acc4ce03f52e8d4c812a3141cbac/connect-demo/jdbc_smt_fromjson.json) is an example that uses `StringConverter` followed by this SMT in a `JdbcSinkConnector`.

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/flex019/uploads/confluentcommunity/original/1X/c49438c90c9df282e9996fdf6971be890c71b65a.svg) [@system](https://forum.confluent.io/u/system)\
**Post date:** [30 May 2025 15:26 UTC](https://forum.confluent.io/t/mqtt-source-connector-byte-array-transform-to-json/37690/7 "2025-05-30T15:26:17Z")

</div>

Closed on 30 May 2025 15:26 UTC

This topic was automatically closed 30 days after the last reply. New replies are no longer allowed.