# MQTT Source Connector for Confluent Cloud The fully managed MQTT Source connector for Confluent Cloud attaches to an MQTT broker, subscribes to specified topics, and streams data from these topics into Apache Kafka®. #### NOTE * This Quick Start is for the fully managed Confluent Cloud connector. If you are installing the connector locally for Confluent Platform, see [MQTT Source Connector for Confluent Platform](https://docs.confluent.io/kafka-connectors/mqtt/current/mqtt-source-connector/). * If you require private networking for fully managed connectors, make sure to set up the proper networking beforehand. For more information, see [Manage Networking for Confluent Cloud Connectors](networking/internet-resource.md#clusters-connect-cloud). ## Features The MQTT Source connector provides the following features: * **Topics created automatically**: The connector can automatically create the Kafka topic. * **Multiple tasks**: The connector supports multiple tasks. More tasks may improve performance. * **SSL support**: Supports two-way SSL. * **MQTT client version**: The connector uses `paho mqttv3 client`. For more information and examples to use with the Confluent Cloud API for Connect, see the [Confluent Cloud API for Connect Usage Examples](connect-api-section.md#ccloud-connect-api) section. ## Limitations Be sure to review the following information. ### At least once delivery To support at-least-once semantics, you should run the connector with `mqtt.clean.session.enabled` set to `false`. Under this mode, the connector will assign each task a unique client ID, which will be used for a persistent connection. The topics are distributed across tasks. Confluent recommends you do not change the topic assignments–that is, do not alter the task count, or change the topic listing once the connector is up and running. This can potentially lead to data loss as the connector will be restarted with a different topic to task assignments, and hence, a different client ID. ### Others * **No Schema Registry support**: The connector writes data as raw bytes using `ByteArrayConverter` and does not support Confluent Cloud Schema Registry or Schema Registry-based formats such as Avro, JSON_SR, or Protobuf. For details and alternatives, see [Why can’t I use Confluent Cloud Schema Registry or SMTs with this connector?](#cc-mqtt-source-faq-sr-smt). * **No managed Single Message Transformations (SMTs) support**: Because the connector writes raw bytes, the managed (out-of-the-box) SMTs that Confluent provides cannot operate on the data. You can still use a [custom SMT](configure-custom-single-message-transforms/quick-start-custom-smt.md#cc-custom-single-message-transforms) that is written to accept and transform bytes. For details and alternatives, see [Why can’t I use Confluent Cloud Schema Registry or SMTs with this connector?](#cc-mqtt-source-faq-sr-smt). * For connector limitations, see [MQTT Source Connector](limits.md#cc-mqtt-source-limits). ## Quick Start Use this quick start to get up and running with the Confluent Cloud MQTT source connector. The quick start shows how to attach the connector to an MQTT broker, subscribe to the specified topics, and stream data into Apache Kafka®. Prerequisites : - Authorized access to a [Confluent Cloud](https://www.confluent.io/confluent-cloud/) cluster on Amazon Web Services (AWS), Microsoft Azure (Azure), or Google Cloud. - Access to an MQTT broker. - The Confluent CLI installed and configured for the cluster. See [Install the Confluent CLI](https://docs.confluent.io/confluent-cli/current/install.html). - The connector automatically creates Kafka topics or you can create the topic manually. - For networking considerations, see [Networking and DNS](overview.md#connect-internet-access-resources). To use a set of public egress IP addresses, see [Public Egress IP Addresses for Confluent Cloud Connectors](static-egress-ip.md#cc-static-egress-ips).
- Kafka cluster credentials. The following lists the different ways you can provide credentials. - Enter an existing [service account](service-account.md#s3-cloud-service-account) resource ID. - Create a Confluent Cloud [service account](service-account.md#s3-cloud-service-account) for the connector. Make sure to review the ACL entries required in the [service account documentation](service-account.md#s3-cloud-service-account). Some connectors have specific ACL requirements. - Create a Confluent Cloud API key and secret. To create a key and secret, you can use [confluent api-key create](https://docs.confluent.io/confluent-cli/current/command-reference/api-key/confluent_api-key_create.html) *or* you can autogenerate the API key and secret directly in the Cloud Console when setting up the connector. ### Using the Confluent Cloud Console #### Step 1: Launch your Confluent Cloud cluster To create and launch a Kafka cluster in Confluent Cloud, see [Create a kafka cluster in Confluent Cloud](../get-started/index.md#cloud-create-kafka-cluster). #### Step 2: Add a connector In the left navigation menu, click **Connectors**. If you already have connectors in your cluster, click **+ Add connector**. #### Step 3: Select your connector Click the **MQTT Source** connector card.  #### Step 4: Enter the connector details #### NOTE * Make sure you have all your [prerequisites](#cc-mqtt-source-prereqs) completed. * An asterisk ( \* ) designates a required entry. At the **Add MQTT Source Connector** screen, complete the following: ### Topic selection Select the topic you want to send data to from the **Topics** list. To create a new topic, click **+Add new topic**. ### Kafka access 1. Select the way you want to provide **Kafka Cluster credentials**. You can choose one of the following options: - **My account**: This setting allows your connector to globally access everything that you have access to. With a user account, the connector uses an API key and secret to access the Kafka cluster. This option is not recommended for production. - **Service account**: This setting limits the access for your connector by using a [service account](service-account.md#s3-cloud-service-account). This option is recommended for production. - **Use an existing API key**: This setting allows you to specify an API key and a secret pair. You can use an existing pair or create a new one. This method is not recommended for production environments. #### NOTE Freight clusters support only service accounts for Kafka authentication. 2. Click **Continue**. ### Authentication 1. Configure the authentication properties: **How should we connect to MQTT Broker?** - **List of Server URIs**: The URI of the MQTT broker. This must be provided in the format `//:URI>`. The supported protocols are TCP, SSL, WS, and WSS. For a connection that uses TLS, you must provide the required key stores and trust stores. - **Username**: The username the connector will use to connect to the host. - **Password**: Password to connect with. **Connection Details** - **MQTT Topics**: MQTT topics to subscribe to. **MQTT secure connection** - **SSL Keystore**: The key store containing the server certificate. - **Keystore Password**: Password used to open the Java KeyStore file. - **Key Password**: Password for the client certificate contained in the Java KeyStore. - **SSL Truststore**: The trust store containing server CA certificate. - **Truststore Password**: Password used to open the Java TrustStore file. 2. Click **Continue**. ### Configuration **Connection Details** - **MQTT QOS**: The MQTT QOS level to subscribe to. Valid values are 0, 1, and 2. - **Records Buffer Queue Size**: The capacity of the buffer queue that stores the data before sending it to the Kafka topic. - **Records Buffer Queue Maximum Batch Size**: The maximum size of the batch to send to Kafka topic in a single request. - **Records Buffer Queue Empty Timeout (Milliseconds)**: Timeout to wait on an empty buffer queue for extracting records. - **Records Buffer Queue Full Timeout (Milliseconds)**: Timeout to wait on a full buffer queue for inserting new records. ### **Show advanced configurations** **Auto-restart policy** - **Enable Connector Auto-restart**: Enables the auto-restart behavior of the connector and its task in the event of user-actionable errors. Defaults to `true`, enabling the connector to automatically restart in case of user-actionable errors. Set this property to `false` to disable auto-restart for failed connectors. If disabled, you must manually restart the connector. **Additional Configs** - **Value Converter Decimal Format**: Specifies the `JSON` or `JSON_SR` serialization format for Connect `DECIMAL` logical type values with two allowed literals: `BASE64` to serialize `DECIMAL` logical types as base64 encoded binary data, and `NUMERIC` to serialize `DECIMAL` logical type values in `JSON` or `JSON_SR` as a number representing the decimal value. - **Key Converter Schema ID Serializer**: The class name of the schema ID serializer for keys. This is used to serialize schema IDs in the message headers. - **Value Converter Reference Subject Name Strategy**: Sets the subject reference name strategy for values. Valid entries are `DefaultReferenceSubjectNameStrategy` or `QualifiedReferenceSubjectNameStrategy`. You can use this strategy only with `PROTOBUF` format; the default strategy is `DefaultReferenceSubjectNameStrategy`. - **Value Converter Connect Meta Data**: Enables the Connect converter to add its metadata to the output schema. Applies to Avro converters. - **Value Converter Value Subject Name Strategy**: Determines how to construct the subject name under which the value schema is registered with Schema Registry. - **Key Converter Key Subject Name Strategy**: Determines how to construct the subject name for key schema registration. - **Value Converter Schema ID Serializer**: The class name of the schema ID serializer for values. This is used to serialize schema IDs in the message headers. **Connection Details** - **Clean Session?**: Select whether the client and server should remember state across restarts and reconnects. - **Connection Timeout**: Sets the connection timeout value in seconds. - **Connection Keepalive**: Defines the maximum time interval in seconds between messages sent or received. - **Max Retry Time**: The maximum amount of time in milliseconds (ms) the connector will spend backing off and retrying failed operations (connecting to the MQTT broker and publishing records). - Click **Continue** ### Sizing Based on the number of topic partitions you select, you will be provided with a recommended number of tasks. 1. To change the number of tasks, use the Range Slider to select the desired number of tasks. 2. Click **Continue**. ### Review and Launch 1. Verify the connection details by previewing the running configuration. 2. Once you’ve validated that the properties are configured to your satisfaction, click **Launch**. The status for the connector should go from **Provisioning** to **Running**. #### Step 5: Check the Kafka topic After the connector is running, verify that messages are populating your Kafka topic. For more information and examples to use with the Confluent Cloud API for Connect, see the [Confluent Cloud API for Connect Usage Examples](connect-api-section.md#ccloud-connect-api) section. ### Using the Confluent CLI Complete the following steps to set up and run the connector using the Confluent CLI. #### NOTE Make sure you have all your [prerequisites](#cc-mqtt-source-prereqs) completed. #### Step 1: List the available connectors Enter the following command to list available connectors: ```none confluent connect plugin list ``` #### Step 2: List the connector configuration properties Enter the following command to show the connector configuration properties: ```none confluent connect plugin describe ``` The command output shows the required and optional configuration properties. #### Step 3: Create the connector configuration file Create a JSON file that contains the connector configuration properties. The following example shows the required connector properties. ```none { "connector.class": "MqttSource", "name": "MqttSource_0", "kafka.auth.mode": "KAFKA_API_KEY", "kafka.api.key": "", "kafka.api.secret": "", "kafka.topic" : "data_topic_0", "mqtt.server.uri" : "tcp://192.0.0.1:1881", "mqtt.topics" : "broker_topic_0", "tasks.max" : "1" } ``` Note the following property definitions: * `"name"`: Sets a name for your new connector. * `"connector.class"`: Identifies the connector plugin name. * `"kafka.auth.mode"`: Identifies the connector authentication mode you want to use. There are two options: `SERVICE_ACCOUNT` or `KAFKA_API_KEY` (the default). To use an API key and secret, specify the configuration properties `kafka.api.key` and `kafka.api.secret`, as shown in the example configuration (above). To use a [service account](service-account.md#s3-cloud-service-account), specify the **Resource ID** in the property `kafka.service.account.id=`. To list the available service account resource IDs, use the following command: ```bash confluent iam service-account list ``` For example: ```bash confluent iam service-account list Id | Resource ID | Name | Description +---------+-------------+-------------------+------------------- 123456 | sa-l1r23m | sa-1 | Service account 1 789101 | sa-l4d56p | sa-2 | Service account 2 ``` * `"kafka.topic"`: The Kafka topic name where you want data sent. If not used, the default topic name created is `mqtt`. * `"mqtt.server.uri"`: The MQTT broker URI. Must be in the format `//:URI`. If the MQTT broker does not support anonymous mode, you must add the following two additional properties: - `"mqtt.username"`: `""` - `"mqtt.password"`: `""` The supported protocols are TCP, SSL, WS, and WSS. For TLS connections, you must supply the keystore and/or truststore file contents and the file passwords creating the connector configuration JSON. The truststore and keystore files are binary files. For the `mqtt.ssl.trust.store.file` and the `mqtt.ssl.key.store.file` properties, you encode the truststore or keystore file in base64, take the encoded string, add the `data:text/plain;base64,` prefix, and then use the entire string as the property entry. For example: ```properties "mqtt.ssl.key.store.file" : "data:text/plain;base64,/u3+7QAAAAIAAAACAAAAAQAGY2xpZ...omitted...==" ``` In addition to adding the file contents, you must also supply the passwords using the properties `mqtt.ssl.key.password`, `mqtt.ssl.key.store.password` and `mqtt.ssl.trust.store.password`. These are the passwords you used while creating the jks and pkcs12 keystore and truststore. For all property values and definitions, see [Configuration Properties](#cc-mqtt-source-config-properties). * `"mqtt.topics"`: The broker topic (or comma-separated broker topics) to subscribe to. * `"tasks.max"`: Enter the number of [tasks](/platform/current/connect/concepts.html#tasks) in use by the connector. The connector supports multiple tasks. More tasks may improve performance. #### NOTE Because this connector writes data as raw bytes using `ByteArrayConverter`, it does not support Confluent Cloud Schema Registry or the managed (out-of-the-box) SMTs that Confluent provides. You can still use a [custom SMT](configure-custom-single-message-transforms/quick-start-custom-smt.md#cc-custom-single-message-transforms) that is written to operate on bytes data. For more information, see [Why can’t I use Confluent Cloud Schema Registry or SMTs with this connector?](#cc-mqtt-source-faq-sr-smt). #### Step 4: Load the properties file and create the connector Enter the following command to load the configuration and start the connector: ```none confluent connect cluster create --config-file .json ``` For example: ```none confluent connect cluster create --config-file mqtt-source.json ``` Example output: ```none Created connector MqttSource_0 lcc-ix4dl ``` #### Step 5: Check the connector status Enter the following command to check the connector status: ```none confluent connect plugin list ``` Example output: ```none ID | Name | Status | Type +-----------+-----------------+---------+-------+ lcc-ix4dl | MqttSource_0 | RUNNING | source ``` #### Step 6: Check the results on the broker. After the connector is running, verify that messages are populating your Kafka topic. For more information and examples to use with the Confluent Cloud API for Connect, see the [Confluent Cloud API for Connect Usage Examples](connect-api-section.md#ccloud-connect-api) section. ## Configuration Properties Use the following configuration properties with the fully managed connector. For self-managed connector property definitions and other details, see the connector docs in [Self-managed connectors for Confluent Platform](/platform/current/connect/kafka_connectors.html). ### How should we connect to your data? `name` : Sets a name for your connector.
* Type: string * Valid Values: A string at most 64 characters long * Importance: high ### Kafka Cluster credentials `kafka.auth.mode` : Kafka Authentication mode. It can be one of KAFKA_API_KEY or SERVICE_ACCOUNT. It defaults to KAFKA_API_KEY mode, whenever possible.
* Type: string * Valid Values: SERVICE_ACCOUNT, KAFKA_API_KEY * Importance: high `kafka.api.key` : Kafka API Key. Required when kafka.auth.mode==KAFKA_API_KEY.
* Type: password * Importance: high `kafka.service.account.id` : The Service Account that will be used to generate the API keys to communicate with Kafka Cluster.
* Type: string * Importance: high `kafka.api.secret` : Secret associated with Kafka API key. Required when kafka.auth.mode==KAFKA_API_KEY.
* Type: password * Importance: high ### Which topic do you want to send data to? `kafka.topic` : Identifies the topic name to write the data to.
* Type: string * Importance: high ### How should we connect to MQTT Broker? `mqtt.server.uri` : The URI of the MQTT broker. This must be given in the format //:URI. The supported protocols are tcp, ssl, ws, wss. Note that for a connection that uses TLS, you must provide the required key stores and trust stores.
* Type: list * Importance: high `mqtt.username` : Username to connect with, or blank if a username is not required. Note: username field is masked as it may contain sensitive information
* Type: password * Importance: high `mqtt.password` : Password to connect with, or blank if a password is not required.
* Type: password * Default: [hidden] * Importance: high ### MQTT secure connection `mqtt.ssl.key.store.file` : The location of the Java KeyStore file containing the private key to use for authenticating with the server.
* Type: password * Default: [hidden] * Importance: low `mqtt.ssl.key.store.password` : Password used to open the Java KeyStore file.
* Type: password * Default: [hidden] * Importance: medium `mqtt.ssl.key.password` : Password for the client certificate contained in the Java KeyStore.
* Type: password * Default: [hidden] * Importance: high `mqtt.ssl.trust.store.file` : The location of the Java TrustStore file containing the certificates required to validate the SSL connection to the server.
* Type: password * Default: [hidden] * Importance: medium `mqtt.ssl.trust.store.password` : Password used to open the Java TrustStore file.
* Type: password * Default: [hidden] * Importance: medium ### Connection Details `mqtt.clean.session.enabled` : Sets whether the client and server should remember state across restarts and reconnects. Note that for unreceived messages to be received after reconnect you should set the QOS to 1 or above.
* Type: boolean * Default: false * Importance: medium `mqtt.connect.timeout.seconds` : Sets the connection timeout value in seconds.
* Type: int * Default: 30 * Importance: medium `mqtt.keepalive.interval.seconds` : This value, measured in seconds, defines the maximum time interval between messages sent or received. In the absence of a data-related message during the time period, the client sends a very small “ping” message, which the server will acknowledge.
* Type: int * Default: 60 * Importance: medium `max.retry.time.ms` : The maximum time in milliseconds (ms) the connector will spend backing off and retrying failed operations (connecting to the MQTT broker and publishing records).
* Type: int * Default: 30000 (30 seconds) * Importance: medium `mqtt.topics` : The MQTT topics to subscribe to.
* Type: list * Importance: high `mqtt.qos` : The MQTT QOS level to subscribe to. Valid values are 0, 1 and 2.
* Type: int * Default: 0 * Importance: low `records.buffer.queue.size` : The capacity of the buffer queue that stores the data before sending it to the Kafka Topic.
* Type: int * Default: 50000 * Valid Values: [10000,…,100000] * Importance: low `records.buffer.queue.max.batch.size` : The maximum size of the batch to send to Kafka topic in a single request.
* Type: int * Default: 4096 * Valid Values: [100,…,4096] * Importance: low `records.buffer.queue.empty.timeout` : Timeout to wait on an empty buffer queue for extracting records.
* Type: int * Default: 100 * Valid Values: [10,…,10000] * Importance: low `records.buffer.queue.full.timeout` : Timeout to wait on a full buffer queue for inserting new records.
* Type: int * Default: 60000 * Valid Values: [30000,…,300000] * Importance: low ### Number of tasks for this connector `tasks.max` : Maximum number of tasks for the connector.
* Type: int * Valid Values: [1,…] * Importance: high ### Auto-restart policy `auto.restart.on.user.error` : Enable connector to automatically restart on user-actionable errors.
* Type: boolean * Default: true * Importance: medium ### Additional Configs `header.converter` : The converter class for the headers. This is used to serialize and deserialize the headers of the messages.
* Type: string * Importance: low `producer.override.compression.type` : The compression type for all data generated by the producer. Valid values are none, gzip, snappy, lz4, and zstd.
* Type: string * Importance: low `value.converter.allow.optional.map.keys` : Allow optional string map key when converting from Connect Schema to Avro Schema. Applicable for Avro Converters.
* Type: boolean * Importance: low `value.converter.auto.register.schemas` : Specify if the Serializer should attempt to register the Schema.
* Type: boolean * Importance: low `value.converter.connect.meta.data` : Allow the Connect converter to add its metadata to the output schema. Applicable for Avro Converters.
* Type: boolean * Importance: low `value.converter.enhanced.avro.schema.support` : Enable enhanced schema support to preserve package information and Enums. Applicable for Avro Converters.
* Type: boolean * Importance: low `value.converter.enhanced.protobuf.schema.support` : Enable enhanced schema support to preserve package information. Applicable for Protobuf Converters.
* Type: boolean * Importance: low `value.converter.flatten.unions` : Whether to flatten unions (oneofs). Applicable for Protobuf Converters.
* Type: boolean * Importance: low `value.converter.generate.index.for.unions` : Whether to generate an index suffix for unions. Applicable for Protobuf Converters.
* Type: boolean * Importance: low `value.converter.generate.struct.for.nulls` : Whether to generate a struct variable for null values. Applicable for Protobuf Converters.
* Type: boolean * Importance: low `value.converter.int.for.enums` : Whether to represent enums as integers. Applicable for Protobuf Converters.
* Type: boolean * Importance: low `value.converter.latest.compatibility.strict` : Verify latest subject version is backward compatible when use.latest.version is true.
* Type: boolean * Importance: low `value.converter.object.additional.properties` : Whether to allow additional properties for object schemas. Applicable for JSON_SR Converters.
* Type: boolean * Importance: low `value.converter.optional.for.nullables` : Whether nullable fields should be specified with an optional label. Applicable for Protobuf Converters.
* Type: boolean * Importance: low `value.converter.optional.for.proto2` : Whether proto2 optionals are supported. Applicable for Protobuf Converters.
* Type: boolean * Importance: low `value.converter.use.latest.version` : Use latest version of schema in subject for serialization when auto.register.schemas is false.
* Type: boolean * Importance: low `value.converter.use.optional.for.nonrequired` : Whether to set non-required properties to be optional. Applicable for JSON_SR Converters.
* Type: boolean * Importance: low `value.converter.wrapper.for.nullables` : Whether nullable fields should use primitive wrapper messages. Applicable for Protobuf Converters.
* Type: boolean * Importance: low `value.converter.wrapper.for.raw.primitives` : Whether a wrapper message should be interpreted as a raw primitive at root level. Applicable for Protobuf Converters.
* Type: boolean * Importance: low `key.converter.key.schema.id.serializer` : The class name of the schema ID serializer for keys. This is used to serialize schema IDs in the message headers.
* Type: string * Default: io.confluent.kafka.serializers.schema.id.PrefixSchemaIdSerializer * Importance: low `key.converter.key.subject.name.strategy` : How to construct the subject name for key schema registration.
* Type: string * Default: TopicNameStrategy * Importance: low `value.converter.decimal.format` : Specify the JSON/JSON_SR serialization format for Connect DECIMAL logical type values with two allowed literals:
BASE64 to serialize DECIMAL logical types as base64 encoded binary data and
NUMERIC to serialize Connect DECIMAL logical type values in JSON/JSON_SR as a number representing the decimal value.
* Type: string * Default: BASE64 * Importance: low `value.converter.flatten.singleton.unions` : Whether to flatten singleton unions. Applicable for Avro and JSON_SR Converters.
* Type: boolean * Default: false * Importance: low `value.converter.reference.subject.name.strategy` : Set the subject reference name strategy for value. Valid entries are DefaultReferenceSubjectNameStrategy or QualifiedReferenceSubjectNameStrategy. Note that the subject reference name strategy can be selected only for PROTOBUF format with the default strategy being DefaultReferenceSubjectNameStrategy.
* Type: string * Default: DefaultReferenceSubjectNameStrategy * Importance: low `value.converter.value.schema.id.serializer` : The class name of the schema ID serializer for values. This is used to serialize schema IDs in the message headers.
* Type: string * Default: io.confluent.kafka.serializers.schema.id.PrefixSchemaIdSerializer * Importance: low `value.converter.value.subject.name.strategy` : Determines how to construct the subject name under which the value schema is registered with Schema Registry.
* Type: string * Default: TopicNameStrategy * Importance: low ## Frequently asked questions Find answers to frequently asked questions about the MQTT Source connector for Confluent Cloud. ### Why is my connector failing to connect to the MQTT broker? Connection failures can occur for several reasons: * **Incorrect broker URI format**: Ensure your `mqtt.server.uri` is in the correct format. Supported protocols are `tcp`, `ssl`, `ws`, and `wss`. * **Authentication required**: If your MQTT broker does not support anonymous mode, provide `mqtt.username` and `mqtt.password` in your connector configuration. * **Network access**: Verify that the connector can reach your MQTT broker. Check firewall rules, VPC peering configurations, or private networking requirements. * **Connection timeout**: Adjust the `mqtt.connect.timeout.seconds` property (default: 30 seconds) if your broker requires more time to establish connections. ### Why does the connector fail with authorization errors when using SSL/TLS? This error typically occurs when the connector is not properly configured to send client certificates during the SSL handshake. To resolve this issue: * **Verify certificate configuration**: Ensure that both the keystore and truststore files are configured with the following properties: - `mqtt.ssl.key.store.file`: Contains your client certificate. - `mqtt.ssl.trust.store.file`: Contains the CA certificate chain. - `mqtt.ssl.key.store.password`: Password for the keystore. - `mqtt.ssl.trust.store.password`: Password for the truststore. * **Use correct base64 encoding**: Ensure the keystore and truststore files are base64-encoded and prefixed with `data:text/plain;base64`. * **Verify MQTT broker configuration**: Ensure that your MQTT broker is configured to require and accept client certificate authentication. ### Why can’t I use Confluent Cloud Schema Registry or SMTs with this connector? The MQTT Source connector for Confluent Cloud writes both keys and values as bytes using `ByteArrayConverter`. This design limitation means: * **No Schema Registry support**: The connector cannot automatically extract, validate, or register schemas in Confluent Cloud Schema Registry. All data is written as raw bytes without schema metadata. * **No managed SMT support**: The managed (out-of-the-box) SMTs that Confluent provides—such as `Filter`, `ExtractField`, or `Cast`—require structured data like Avro, JSON Schema, or Protobuf and cannot operate on byte arrays. Configuring a managed SMT with this connector can result in configuration or runtime errors. You can still use a [custom SMT](configure-custom-single-message-transforms/quick-start-custom-smt.md#cc-custom-single-message-transforms) if it is written to accept and operate on bytes data. **Alternative approaches**: Since the connector itself cannot use Confluent Cloud Schema Registry or managed SMTs, process the data with one of the following approaches: * **For schema validation and registration**: Use ksqlDB, Confluent Cloud for Apache Flink, or Kafka Streams to deserialize the byte data, apply schemas, and write to a new topic with proper schema registration in Confluent Cloud Schema Registry. * **For data transformation**: Apply transformations using a custom SMT or stream processing rather than the managed SMTs: - Deploy a [custom SMT](configure-custom-single-message-transforms/quick-start-custom-smt.md#cc-custom-single-message-transforms) that is written to accept and operate on the bytes data the connector produces. - Use ksqlDB to filter, transform, or enrich the data with SQL-like queries. - Use Confluent Cloud for Apache Flink to create stateless or stateful transformations with a visual interface. - Use Kafka Streams applications for custom transformation logic. * **For consumer applications**: Implement custom deserialization logic in your consumer applications to parse the byte data according to your MQTT message payload format (for example, JSON, Protobuf, or custom binary formats). ### Why am I experiencing data loss or low throughput with the connector? Data loss or unexpectedly low throughput can occur due to several factors. **To prevent data loss**: * **Disable clean session**: Set `mqtt.clean.session.enabled` to `false` to ensure at-least-once delivery semantics. This enables the connector to maintain persistent connections with unique client IDs per task. * **Avoid changing task count**: Do not change the `tasks.max` value or modify the `mqtt.topics` subscription once the connector is running. Changes to task count redistribute topics across tasks with different client IDs, potentially causing message loss. * **Use appropriate QoS level**: Configure your MQTT broker to use Quality of Service (QoS) level one or two for guaranteed message delivery. **To improve the connector throughput**: * **Increase task count**: Use the `tasks.max` property to run multiple tasks in parallel, distributing the load across tasks. * **Optimize MQTT topics**: Distribute your data across multiple MQTT topics to allow better parallelization across connector tasks. * **Tune buffer queue parameters**: Adjust the following configuration properties to optimize throughput: - `records.buffer.queue.size`: Sets the maximum number of records the buffer queue can hold. - `records.buffer.queue.max.batch.size`: Controls the maximum batch size for records processed from the queue. - `records.buffer.queue.empty.timeout`: Specifies how long to wait when the queue is empty before processing. - `records.buffer.queue.full.timeout`: Specifies how long to wait when the queue is full before blocking. ### How do I troubleshoot connection failures? Connection failures typically result from network issues or incorrect MQTT broker configuration. To diagnose and resolve connection problems: 1. **Verify the broker URI**: Ensure the `mqtt.server.uri` property uses the correct format. Supported protocols are: - `tcp://` for non-encrypted connections. - `ssl://` for TLS connections. - `ws://` for WebSocket connections. - `wss://` for secure WebSocket connections. 2. **Test connectivity**: Use a client tool such as `mosquitto_sub` to verify you can connect to the MQTT broker from outside Confluent Cloud: ```bash mosquitto_sub -h your-broker-host -p 1883 -t "test/topic" -v ``` 3. **Check authentication**: If your MQTT broker requires authentication, ensure you have configured the correct `mqtt.username` and `mqtt.password` in the connector properties. 4. **Network access**: Verify that Confluent Cloud can reach your MQTT broker: - For internet-accessible brokers, ensure the broker allows connections from Confluent Cloud IP addresses. - For private networks, configure appropriate networking such as VPC peering, Private Link, or Transit Gateway. 5. **Adjust connection timeout**: If your broker requires more time to establish connections, increase the `mqtt.connect.timeout.seconds` property (default: 30 seconds). 6. **Check DNS resolution**: Transient DNS failures can cause connection errors. Ensure your MQTT broker hostname resolves correctly and consistently. ### What happens if I change the task count or topic subscription after the connector starts? Changing the task count (`tasks.max`) or modifying the MQTT topics subscription (`mqtt.topics`) after the connector is running can lead to data loss. When `mqtt.clean.session.enabled` is set to `false` which is recommended for at-least-once delivery, each task is assigned a unique client ID based on the task number. The MQTT broker maintains separate message queues for each client ID. If you change the task count: * Topics are redistributed across different tasks. * The mapping of MQTT topics to tasks is altered. * Client IDs for the tasks remain the same, but the previous state associated with the old topic-to-task mapping is lost. Note: If `mqtt.clean.session.enabled` is set to `true`, new client IDs are generated for the tasks, and messages queued for the old client IDs are not consumed by the new client IDs. This results in data loss for messages received between the configuration change **Best practices**: * Finalize your `tasks.max` and `mqtt.topics` configuration before starting the connector. * If changes are necessary, consider creating a new connector with a different name instead of reconfiguring the existing one. ## Next Steps For an example that shows fully managed Confluent Cloud connectors in action with Confluent Cloud for Apache Flink, see the [Cloud ETL Demo](/platform/current/tutorials/examples/cloud-etl/docs/index.html). This example also shows how to use Confluent CLI to manage your resources in Confluent Cloud. [](https://docs.confluent.io/platform/current/tutorials/examples/cloud-etl/docs/index.html)