Release Notes
Cluster Version Updates
Product Announcements
Message Gaps
IndividualDeletedMessages is part of the subscription progress in the Shared and Key_Shared modes in open-source Apache Pulsar, and is generally referred to as a message gap.
We all know that in Apache Pulsar, consumption progress consists of two parts:
MarkDeletePosition: The initial consumption position. Messages at or before this position are considered consumed. This is equivalent to the offset in Kafka.
IndividualDeletedMessages: The set of acknowledged messages, also known as message gaps. It refers to the set of positions of acknowledged messages after the MarkDeletePosition.
Subscription Progress Concepts
Take the message consumption progress diagram as an example:
MarkDeletePosition = 1:2
IndividualDeletedMessages = [ (1:4 - 1:6], (1:7 - 1:8] ]
There are many situations where message gaps occur, such as discontinuous acknowledged messages. The common ones are as follows:
1. In the Shared or Key_Shared consumption mode, some messages are consumed quickly while others are consumed slowly, which easily leads to message gaps.
2. In a topic with delayed messages, messages at positions such as 1:3, 1:4, and 1:7 in the diagram may not have reached their delay time, while messages at positions like 1:5, 1:6, and 1:8 have already reached their delay time. In this case, gaps will also occur.
Difference Between Message Gaps and Unack
Unack indicates that the server has pushed the message to the consumer's in-memory queue, but the consumer has not returned an acknowledgment to the server, that is, the consumer has not called the APIs related to consumer.acknowledge. Common scenarios include unacknowledged messages caused by slow consumption or blocked consumption logic.
For example, in the message consumption progress diagram, messages 1:4 and 1:10 are those pushed to the consumer by the server but not acknowledged by the consumer. These messages will be counted as unacknowledged messages.
Note:
In summary, message gaps and unack are distinct concepts with no direct relationship. However, unack can cause message gaps in certain scenarios, such as those shown in the message consumption progress diagram.
Quantity Limit Background
The IndividualDeletedMessages part in the subscription progress needs to be persisted periodically. During the persistence process, the IndividualDeletedMessages object is serialized into a string and stored as an internal message in the Bookie cluster. This ensures that the complete subscription progress information can be retrieved after the topic is reloaded.
In Apache Pulsar, the number of stored IndividualDeletedMessages is limited by the cluster-level configuration item managedLedgerMaxUnackedRangesToPersist, which defaults to 10000. When the number of IndividualDeletedMessages exceeds 10000, only the first 10000 entries in IndividualDeletedMessages will be persisted, and the part exceeding 10000 will be lost.
Note:
Impact of incomplete persistence of IndividualDeletedMessages: The consumption progress will be incomplete. If a topic is unexpectedly reloaded (such as topic unload or broker restart), some messages may be consumed repeatedly.
It is recommended that you configure gap message alarms for instance topics. Set the relevant threshold based on actual conditions, and it is recommended to set it between 8,000 and 10,000 at least. This ensures that the number of message gaps does not exceed the upper limit. For configuration details, see Configuring Alarms
The reasons for the limit can be primarily attributed to the following two aspects:
1. Without a limit, if the number of IndividualDeletedMessages becomes too large, the length of internal messages will increase. Message queues are not well-suited for handling excessively large messages. In this case, messages generated by the internal subscription progress would impose excessive pressure on the cluster and may even affect the stability of the entire cluster.
2. IndividualDeletedMessages are also stored as messages in the bookie and need to comply with the maximum message size limit (5 MB). Based on previous test experience, when the number of IndividualDeletedMessages exceeds 200,000, the generated message length reaches 5 MB.
Generation of Message Gaps and Use Cases
IndividualDeletedMessages can be generated in the following 4 scenarios. Among them, delayed messages and Key_Shared subscriptions may produce a large number of IndividualDeletedMessages, and in these two scenarios, the risk of progress loss is higher.
Scenario 1: Unacknowledged Messages
Subscription mode: Shared and Key_Shared
Subscription progress:
MarkDeletePosition = 1:2
IndividualDeletedMessages = [ (1:4 - 1:6], (1:7 - 1:8] ]
The generation of IndividualDeletedMessages is related to consumption behavior, and the consumption time of different messages varies.
When the server pushes messages to the client, the time required to consume each message varies. Messages produced later might be consumed first. In such cases, this results in an increase in the number of IndividualDeletedMessages.
IndividualDeletedMessages will not exceed the maximum number of unacknowledged messages (default: 5000).
Conclusion: The number of IndividualDeletedMessages cannot exceed the maximum number of unacknowledged messages (default: 5000). Generally, there is no risk that the number of IndividualDeletedMessages exceeds the limit specified by managedLedgerMaxUnackedRangesToPersist.
Scenario 2: Delayed Messages
Subscription mode: Shared and Key_Shared
Subscription progress:
MarkDeletePosition = 1:2
IndividualDeletedMessages = [ (1:4 - 1:6], (1:7 - 1:8] ]
The number of IndividualDeletedMessages is related to the distribution of delayed messages in the queue.
When the server reads a message, if it identifies the message as a delayed message and the delay time has not been reached, the message at the corresponding position cannot be pushed to the consumer. When the delay time of subsequent messages is reached earlier than that of previously sent messages, the later-produced messages will be consumed first. In this case, the number of IndividualDeletedMessages will increase.
If the number of message gaps exceeds 10,000, the subscription progress may be partially lost when the topic is reloaded.
Conclusion:The number of IndividualDeletedMessages may exceed the limit specified by managedLedgerMaxUnackedRangesToPersist.
Use Cases
How to reduce (control) the number of message gaps? Based on the above explanation of how IndividualDeletedMessages are generated, the following suggestions can typically decrease the number of message gaps or avoid duplicate messages caused by message gaps:
1. Consumers should have implemented idempotence as a fallback solution.
2. Place delayed messages and non-delayed messages into separate topics.
3. Try to keep the delay time of delayed messages in a certain sequentiality (preferably in ascending order). This practice needs to be combined with actual business scenarios.
4. Try to group messages with the same delay time together. This practice needs to be combined with actual business scenarios.
5. Scale out partitions and ensure data balance across partitions.
Optimization Solutions
Considering that there may be deviations in understanding based on actual business scenarios, we provide two solutions. These solutions specifically target maintaining the sequentiality of delayed messages and integrating the sending of messages with the same delay time. The following optimization solutions require modifications to the business system.
Optimization solution 1: Keep the delay time of delayed messages in a certain sequentiality (like the time wheel mechanism).
Delayed messages with a delay time within 1 day are delivered to a random delay time topic based on their actual delay time; delayed messages with a delay time exceeding 1 day are delivered to a fixed delay time topic based on a fixed delay time.
The business simultaneously subscribes to 2 topics. When a message is consumed and its delay time has not reached the actual delay time specified by the user, redeliver the message according to the above logic (messages with a delay time within 1 day are delivered to the random delay time topic, and messages with a delay time exceeding 1 day are delivered to the fixed delay time topic).
Message Sending Stage
Determine the topic to send based on the expiration time of messages.
Deliver delayed messages with a delay time within 1 day to Topic 1. (A random delay-time topic. The delay time of messages in this topic is set based on their actual delay time.)
Deliver delayed messages with a delay time exceeding 1 day to Topic 2. (A fixed delay time topic. The delay time of messages in this topic is fixed at 1 day.) The actual delay time is stored in the message properties.
Consumption Stage
Topic 1 and Topic 2 both have consumers.
Consumers directly consume delayed messages in Topic 1 after the message delay time expires.
Consumers consume delayed messages in Topic 2 after the message delay time expires, based on the actual delay time obtained from the message properties. If the actual delay time is less than 1 day, messages are delivered to Topic 1 based on the actual expiration time; if the actual expiration time exceeds 1 day, messages continue to be delivered to Topic 2 with a fixed 1-day delay time.
Benefits of the Solution
The delayed messages in Topic 1 are managed in a 1-day time range, effectively controlling the number of message gaps.
The delayed messages in Topic 2 are set with a fixed delay time, ensuring the expiration times remain ordered and avoiding the message gap issue that occurs in scenarios with random delay times.
Disadvantages
1. Multiple rounds of consumption and redelivery (consumption and resending) are required for delayed messages with an actual delay time exceeding 1 day. Theoretically, the maximum number of redeliveries required is equal to the actual number of delay days.
How to reduce the number of redeliveries for delayed messages caused by the above solution?
This issue can be resolved by using a higher-level time wheel approach. For example, as shown in the figure below, add a topic with a fixed delay time of 5 days. This can reduce the number of redeliveries for messages with an actual delay time exceeding 5 days.
2. If the number of message gaps caused by delayed messages within 1 day is still large, the time range of the random delay time topic needs to be further shortened (for example, to 12 hours). This adjustment will increase the number of message redeliveries.
Optimization Solution 1: Keeping the Delay Time of Delayed Messages in a Certain Sequentiality (Like the Time Wheel Mechanism)
Split into multiple topics, where each topic stores delayed messages that expire within a specific day.
Message Sending Stage
Determine the topic to send based on the expiration time of messages. Example:
Messages expiring on January 1st will be sent to Topic 1.
Messages expiring on January 2nd will be sent to Topic 2.
Messages expiring on January 3rd will be sent to Topic 3.
Consumption Stage
Each topic has its own consumers, and each consumer only consumes the expired messages from the topic it is responsible for.
Messages expiring on January 1st will ultimately be consumed by consumers in Topic 1 once they expire.
Messages expiring on January 2nd will ultimately be consumed by consumers under Topic 2 once they expire.
Messages expiring on January 3rd will ultimately be consumed by consumers under Topic 3 once they expire.
Benefits of the Solution
By controlling the delay time range of delayed messages in a single topic, this solution ensures that the delayed messages in each topic expire on the same day. This way, message gaps will only be concentrated within a one-day range, effectively limiting the number of message gaps.
Disadvantages
1. This solution requires managing more topics and maintaining message transfer relationships between them.
2. If the number of message gaps caused by delayed messages within 1 day is still large, topics need to be further split. However, this may involve managing more topics.
Scenario 3: Key_Shared Subscription
Subscription mode: Key_Shared
Subscription progress:
MarkDeletePosition = 1:2
IndividualDeletedMessages = [ (1:4 - 1:6], (1:7 - 1:8] ]
The number of IndividualDeletedMessages is related to the message push status.
When the number of messages under certain keys is too large, or their consumption is too slow, the consumer cannot receive more messages corresponding to those keys. After the server reads the messages of the corresponding keys, it cannot push them to the client. If subsequent messages can be delivered to the consumer and are consumed before the earlier messages, the number of IndividualDeletedMessages will increase.
If the number of message gaps exceeds 10,000, the subscription progress may be partially lost when the topic is reloaded.
Conclusion:The number of IndividualDeletedMessages may exceed the limit specified by managedLedgerMaxUnackedRangesToPersist.
Use Cases
1. Ensure that consumers implement idempotence.
2. Ensure rapid enhancement of consumption capacity (horizontal scale-out capability) to proactively prevent message backlogs. This aligns with the official documentation's use cases for the Key_Shared mode.
3. A large number of keys with a relatively even distribution of messages across each key. This avoids message backlogs caused by data skew or slow processing of some keys.
Scenario 4: Exclusive/Failover Subscription
Subscription mode: Exclusive and Failover
Subscription progress:
MarkDeletePosition = 1:2
IndividualDeletedMessages = [ (1:4 - 1:6], (1:7 - 1:8] ]
The difference between this scenario and Scenario 1: Unacknowledged Messages is that the server does not count unacknowledged messages in Exclusive and Failover modes. As long as the consumer receives a message, the server can continue pushing subsequent messages. Therefore, if some messages are not acknowledged, the number of IndividualDeletedMessages will increase.
Conclusion: The number of IndividualDeletedMessages may exceed the limit specified by managedLedgerMaxUnackedRangesToPersist.
Use Cases
1. It is recommended to use the cumulative acknowledgment (acknowledgeCumulative) method to avoid message gaps (IndividualDeletedMessages).
2. If you are using the single-message acknowledgment (acknowledge) method, follow the guidelines below:
Ensure that consumers implement idempotence.
Ensure the balanced distribution of message volumes across partitions.
Ensure rapid enhancement of consumption capacity (horizontal scaling capability) to proactively prevent message backlogs.
3. For the Failover mode, scaling out partitions may alleviate the issue.
Help and Support
Was this page helpful?
You can also Contact sales or Submit a Ticket for help.
Help us improve! Rate your documentation experience in 5 mins.
Feedback