Topic and Subscription

Tuya IoT message queue is a message middleware that is deeply customized based on the open source message component Apache Pulsar and adapted to Tuya’s business scenarios. It retains many features of Pulsar.
A topic is the subject of production and consumption in the Tuya IoT message queue. Subscription is equivalent to a consumer group under a topic. In the same subscription, a message will only be consumed and acknowledged once by a consumer. After you enable the Message Service in your cloud project, the system automatically creates a topic and a default subscription.

Pulsar architecture

A Pulsar cluster consists of at least three parts:

• ZK, the metadata storage and configuration center

• Bookie, the persistent storage of messages

• Broker, the stateless proxy component

  

Topic and subscription format

Topic format

{persistent|non-persistent}://tenant/namespace/topic

• All topics in the message service are persistent topics. Messages are persistently stored to disk until they are acknowledged or expire.
• tenant: Defaults to the accessId/clientId of the current cloud project.
• namespace: Defaults to out.
• topic: The production channel topic is event. It is also a partitioned topic with 11 partitions. The test channel topic is event-test, which is a non-partitioned topic.

Subscription format

tenant-suffix

• tenant: Defaults to the accessId/clientId of the current cloud project.
• suffix: The suffix of the subscription name. The default value is sub.

Topic partitions

Topics in Pulsar are divided into partitioned topics and non-partitioned topics.

• Partitioned topic: refers to a topic with multiple partitions. When a producer pushes messages, it uses a business ID (such as a device ID) to hash and distribute the messages to different topic partitions. That is to say, messages generated by the same business ID will be pushed to the same partition.
• Non-partitioned topic: Also referred to as a single-partition topic, a topic with only one partition. It is generally used in production and consumption scenarios where messages need to be globally in order, and consumption bottlenecks are prone to occur.

Subscription to Pulsar topics

Subscription types

There are four subscription types in Pulsar: Exclusive, Failover, Shared, and Key_Shared.

• Exclusive: The exclusive type is a subscription type that only allows a single consumer to attach to the subscription. If multiple consumers subscribe to a topic using the same subscription, an error occurs. Note that if the topic is partitioned, all partitions will be consumed by the single consumer allowed to be connected to the subscription.

  The exclusive type is suitable for globally ordered messages. However, this subscription type is not recommended if the consumption capacity is limited.

• Failover: The failover type is a subscription type that multiple consumers can attach to the same subscription. However, at any given time, a master consumer is picked for a non-partitioned topic or each partition of a partitioned topic and receives messages. When the master consumer disconnects, all (non-acknowledged and subsequent) messages are delivered to the next consumer in line.
  For partitioned topics, the broker sorts consumers by priority and lexicographical order of consumer name.

  • The failover type is suitable for scenarios with local orderly consumption.
  • In the scenario of Tuya’s IoT message queue, messages under the same business ID are in order, so the failover subscription type is recommended.
  • You can use it together with acknowledgeCumulative. acknowledgeCumulative is a bulk acknowledgement mechanism. The consumer acknowledges all unacknowledged messages before the message in one operation.

  If the number of consumers is greater than the number of partitions:

  

  If the number of consumers is less than the number of partitions:

  

• Shared: The shared subscription type in Pulsar allows multiple consumers to attach to the same subscription. Messages are delivered in a round-robin distribution across consumers, and any given message is delivered to only one consumer. When a consumer disconnects, all the messages that were sent to it and not acknowledged will be rescheduled for sending to the remaining consumers.

  Shared subscriptions do not guarantee message ordering or support cumulative acknowledgment.

  

• Key_Shared: The Key_Shared subscription type in Pulsar allows multiple consumers to attach to the same subscription. Messages in the Key_Shared type are delivered in distribution across consumers and messages with the same key or same ordering key are delivered to only one consumer. No matter how many times the message is re-delivered, it is delivered to the same consumer.

  • The Key_Shared subscription type ensures that the same key is processed by a single consumer at any given time.

  • When a new consumer is connected and thus added to the list of connected consumers, the algorithm re-adjusts the mapping such that some keys currently mapped to existing consumers will be mapped to the newly added consumer. Meanwhile, the broker will record the position of the newest message, such as X, and associate it with the new consumer. The broker will only start delivering messages to the new consumer after all the messages before X have been acknowledged.

  • Consumers can enable allowOutOfOrderDelivery to disable ordering, thereby ensuring that new consumers can consume messages immediately.

    

  • In the Key_Shared type, you cannot use cumulative acknowledgment.
  • A newly connected consumer may not immediately receive messages until all messages sent before the new consumer is connected are acknowledged.
  • It is not recommended to use Key_Shared because its implementation mechanism is more complicated and might cause stability issues. Topics of the Message Service all have multiple partitions, so it is recommended to use failover mode.

Optimization suggestions for subscription and consumption

The production channels of the Message Service are all partitioned topics, and the default number of partitions is 11.

• The recommended subscription mode is failover, which ensures that messages are consumed in order within a single partition.
• Based on the failover subscription mode, one consumer is sufficient for one partition. That is, starting up to 11 consumers can achieve the maximum consumption speed.
• Based on the failover subscription mode, acknowledgeCumulative is recommended to acknowledge messages. This reduces unexpected scenarios where some messages are not acknowledged, resulting in consumption accumulation.

FAQs

Why does monitoring show backlog, although consumption is in progress?

It is normal to have a small amount of message backlog. When messages are delivered from the producer, Pulsar, and to the consumer, a small amount of messages might not be acknowledged. These messages have not reached the consumer or have not been acknowledged by the consumer. They will be counted as a backlog. Typically, when the message production rate is uniform, a message backlog might occur for milliseconds to several seconds.
A large amount of messages might backlog because the consumption speed cannot keep up. By default, in the production environment, 11 Pulsar consumers can be started to achieve the maximum consumption rate. If there are already more than 11 consumers, you can observe the time consumption of local business processing logic by adding logs and monitoring tools. Find out the bottleneck and modify the consumption code.
If the consumption rate matches the production rate but there is still a backlog, you can try restarting the local consumption service. Alternatively, you can check the consumption logic to see if any scenarios might prevent some messages from being acknowledged.

How to use broadcasting to let all consumers receive the same message?

Pulsar does not support broadcast subscription consumption mode. That is, it is impossible to use the same subscription to allow all consumers to receive the same message. If you need to start multiple consumers and make them receive the same message, it is recommended that you subscribe to receive messages individually. After the consumers receive the message, the distribution logic is implemented in the code.

How to handle a “401/Failed to authenticate” error when starting a consumer?

• Check whether the Pulsar service address of the local consumption service is configured correctly. There are currently four data centers available for the Message Service. Confirm that you have selected the correct Pulsar service address.
• On the Tuya Developer Platform, open the current cloud project and check whether the data center under Message Service is correct. Make sure you have switched to the data center where you want to subscribe to the Pulsar service and enabled the Message Service.
• Check whether the IP allowlist feature is enabled for the cloud project. If the IP allowlist is enabled, confirm the consumer’s IP address has been added to the specified data center. If not enabled, confirm the consumer’s IP address is consistent with the data center.