Use an MQTT source service to read data from an MQ Telemetry Transport (MQTT) broker. To create an MQTT source service, use the MQTT source service type. The MQTT source service type does not include a user name property and a password property. An MQTT source service does not get authenticated by the MQTT broker when it retrieves data from the broker.
If multiple MQTT source services connect to an MQTT broker, each connection must have a unique identifier. The MQTT source service type includes a
Client ID property that you can configure for the connections to the MQTT broker. An MQTT source service needs a persistence mechanism to store messages that it reads from an MQTT broker when they are being processed. The MQTT source service stores the messages in a persistence store, and uses the Client ID as the identifier for the persistence store.
If two MQTT source services have the same client ID, one MQTT source service gets disconnected when the other source connects to the MQTT broker.
In a data flow in which an MQTT source service writes data to a target service, the source service first writes the data to an internal queue.
You can configure the following properties for the MQTT source service type:
- Entity Name
- Name of the MQTT source service. Maximum length
- is 32 characters.
- Description of the source service. Maximum length is 256 characters.
- Server URL
- URL of the MQTT broker or server to which to connect to receive messages. The URL is of the form
- Names of the topics to which to subscribe. The MQTT source service supports the topic wildcards that the MQTT specification describes. You can enter a comma-separated list of topic names.
- Client ID
Optional. Unique identifier that identifies the connection between the MQTT source service and MQTT broker, and the file-based persistence store that the MQTT source service uses to store messages when they are being processed.
Enter a string of any length.
- Maximum Enqueued Messages
- Maximum number of messages that can be stored in the persistence store.
Default is 100000 messages.
- Maximum Messages in a Batch
- Maximum number of messages that are sent by the persistence store in a batch, to a target service.
Default is 100 messages.
- No. of Retries to Add Messages to Persistent Store
- Number of times that the MQTT source service tries to add messages to the internal queue if the internal queue is full.
Default is 3.
- Retry Interval
- The time in milliseconds between retries.
Default is 10 milliseconds.
- Maximum Event Size
- Maximum length of data that the source service can read at a time, in bytes. Default is 8192. Minimum value is 1. Maximum value is 51200.
- Retry on Failure
Indicates that the source service should try to open or read from the source if the operation fails the first time.
- Number of Retries
The number of times the source service should retry to open or read from the source.
- Delay between Retries
- Time in milliseconds between successive retries.
- Persist Data
- Persist the data from the source locally so that the source service can continue to process data even if the target is not available.
- Persistence Options
- You can configure the following persistence options:
- batchSizeInBytes. The batch size of the events in bytes after which the data is written to the persistent store. Specify a value of 0 if you want to write the data to the persistent store immediately. Default is 262144 bytes.
- flushInterval. The interval in milliseconds after which the source services writes data to the persistent store. Default is 5000 milliseconds.
- maxDataFiles. The maximum number of data files that the source service can keep in the persistent store. After the number of files exceed this number, a rollover occurs. Default is 10.
- maxDataFileSize. The maximum size of each data file in bytes. When the file reaches the specified size, a file rollover occurs. Default is 1073741824.
- maxQueueSizeInBytes. The maximum size of data that is unsent or has not received acknowledgment in bytes that can be stored in the persistent store. If the specified size is exceeded, the source service does not read data from the source. If you specify a value of 0, the unsent data can be as much as the value of
maxDataFileSize. Default is unlimited size.
- You can collect and monitor the following statistics for the MQTT source service:
- Bytes Sent. Number of bytes sent by the source service.
- Events Sent. Number of events sent by the source service.
- Events to be Sent. Number of events that the source service is yet to send.
- Events not Delivered. Number of events that the source service did not deliver.
- Send Rate (Per Sec). Number of bytes sent every second.
- Events Dropped. Events dropped by the source service while processing the source data. This statistic increases when the length of the message that is read is greater than the event size.
- Events to be Processed. Maximum number of messages that can be stored in the internal queue of the source.