Key_Shared consumer scale behavior with a partitioned topic

[niriven]

Hi I am curious what happens in a certain situation -- based on some tests I am running locally. I read through https://pulsar.apache.org/docs/3.3.x/concepts-messaging/#preserving-order-of-processing -- but it does not talk much if at all about partitioned topics with key_shared. In fact there is a note here that states When the position of the newest message in a topic is X, a key_shared consumer that is newly attached to the same subscription and connected to the topic will not receive any messages until all the messages before X have been acknowledged. which seems to be inaccurate with partitioned topics and multiple consumers.

Test is as follows

• 1 broker
• 1 persistent topic with two partitions
• One producer application writing to the topic at a rate of 10 sends per second with 10 unique keys, key-N, where N is 0-9.
• I have one subscription created ahead of time in Key_Shared mode with allowOutOfOrderDelivery = false
• I bring Consumer-1 online, it will process all keys (as expected) and acknowledge all except key-0 messages
  • Consumer-1: 13:45:27.718 [pool-1-thread-1] INFO c.t.q.p.c.PulsarSpecifiedKeySkipConsumer1 -IS Received: 210, Acknowledged: 198, Skipped: 12, Errored: 0, Keys in 30s [10]: [key-0, key-1, key-2, key-3, key-4, key-5, key-6, key-7, key-8, key-9]
• I bring Consumer-2 online, Consumer-1 is now only processing 7 of the 10 keys, and Consumer-2 is only processing 1 of the 10 keys
  • Consumer-1: 13:45:51.916 [pool-1-thread-1] INFO c.t.q.p.c.PulsarSpecifiedKeySkipConsumer - Received: 380, Acknowledged: 368, Skipped: 12, Errored: 0, Keys in 30s [7]: [key-1, key-2, key-3, key-4, key-5, key-6, key-7]
  • Consumer-2: 13:45:49.315 [pool-1-thread-1] INFO c.t.q.p.c.PulsarSpecifiedKeySkipConsumer - Received: 30, Acknowledged: 30, Skipped: 0, Errored: 0, Keys in 30s [1]: [key-9]

Based on the test, key-8 remains unaccounted during this period I do not ack the key-0 messages. So based on this test I have a few questions

• I assume when I ack all of the key-0 messages it will re-balance appropriately?
• What is happening here during this time with key-8 not being delivered to any consumer?
• Is there any documentation on how key_shared works with partitioned topics and consumer scale up / down?

NOTE: When I set allowOutOfOrderDelivery to true, it distributes as I'd expect (All consumers are working on all keys during scale up / down)

I have added durable-queue-poc2.zip here with the code that exhibits the behavior. Note that this will write a temp file per sender / receiver instance to keep track of what they are doing. The README.md contains steps including how I started the broker.
durable-queue-poc2.zip

[lhotari]

One important detail with Key_Shared is that on the producer side you will need to use .batcherBuilder(BatcherBuilder.KEY_BASED) or disable batching with .enableBatching(false), otherwise ordering could get mixed (this is also mentioned in docs).
Are you handling this in your test configuration?

[niriven]

@lhotari Thank you for pointing that out. I forgot to set it on the producer, however when I reran the test with enableBatching(false) on the producer I am seeing similar issues.

The new results based on setting enableBatching(false) on the producer

• When Consumer 1 initially comes online, all 10 keys are being received while I am skipping (not acking) all key-0 messages
• When Consumer 2 comes online
  • Consumer 1 08:54:37.714 [pool-1-thread-1] INFO c.t.q.p.c.PulsarSpecifiedKeySkipConsumer - Received: 350, Acknowledged: 299, Skipped: 51, Errored: 0, Keys in 30s [4]: [key-0, key-2, key-3, key-7]
  • Consumer 2 08:54:38.616 [pool-1-thread-1] INFO c.t.q.p.c.PulsarSpecifiedKeySkipConsumer - Received: 100, Acknowledged: 100, Skipped: 0, Errored: 0, Keys in 30s [3]: [key-1, key-5, key-9]

So there is a total of 7 keys being processed across 2 consumers, and 3 unaccounted for (key-4, key-6, key-8)

[lhotari]

So there is a total of 7 keys being processed across 2 consumers, and 3 unaccounted for (key-4, key-6, key-8)

@niriven Thanks for the great repro case and clear instructions. This looks like a bug. I'll investigate it.

[lhotari]

It's pretty much possible that everything is working as expected. Investigating this is also a learning journey for myself.

I assume when I ack all of the key-0 messages it will re-balance appropriately?

If this wouldn't happen, then this would be a bug.

The downside of Key_Shared subscriptions is the limited observability of the subscriptions. It might be hard to detect that delivery is paused because there are too many unacknowledged messages.

There are some tools. partitioned-stats and partitioned-stats-internal create a JSON file with some useful information:

pulsar-admin topics partitioned-stats -etb --per-partition persistent://public/default/my-topic > stats.json
pulsar-admin topics partitioned-stats-internal persistent://public/default/my-topic > stats_internal.json

• stats.json
• stats_internal.json

In stats_internal, you have the individuallyDeletedMessages information which shows the message ranges after markDeletePosition that have already been acknowledged. When the range is continuous, the markDeletePosition will move forward.

In internal stats, the numberOfEntriesSinceFirstNotAckedMessage stat shows how many entries (messages or batch messages) have been added to the topic since the first not acknowledged message. This is about the same information as msgBacklog in stats. unackedMessages information is available in the stats at the consumer level and also at the subscription level. That can be used to find out what is causing the backlog.

What is happening here during this time with key-8 not being delivered to any consumer?

This is a very broad question, but I guess going directly code is a short answer.
When a message cannot be immediately delivered to a consumer, it's added to the MessageRedeliveryController. That keeps track of which messages are pending redelivery and will block the progress of an individual consumer when any of the hash key ranges is pending delivery:

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java

Lines 344 to 349 in 411f697

	if (readType == ReadType.Normal && stickyKeyHashes != null
	&& redeliveryMessages.containsStickyKeyHashes(stickyKeyHashes)) {
	// If redeliveryMessages contains messages that correspond to the same hash as the messages
	// that the dispatcher is trying to send, do not send those messages for order guarantee
	return 0;
	}

When allowOutOfOrderDelivery=true, this logic isn't used.

Is there any documentation on how key_shared works with partitioned topics and consumer scale up / down?

There's probably not much documentation about this. In Pulsar, partitioned topics are no special case. A partitioned topic consists of topic partitions which behave like ordinary topics. Publish messages to partitioned topics explains how publishing works.

The key_shared documentation is a good reference to understand the consumer scale up and scale down.

From the broker's perspective, there's no scale up and scale down, but consumers connecting or disconnecting. The behavior depends on the KeySharedPolicy.autoSplitHashRange() / KeySharedPolicyAutoSplit / KeySharedPolicy.stickyHashRange() / KeySharedPolicySticky defined at the keySharedPolicy of the consumer builder.
The auto split behavior is explained in the key_shared documentation. It also explains the connecting and disconnecting and how this gets handled.

One implicit detail is that the same key could be delivered to a different consumer when consumer connections get disconnected and reconnect. If an application has strict ordering requirements, it's not sufficient to rely on auto split mode.
For strict ordering requirements, it might be better to use partitioned topics and avoid key_shared completely.
If key_shared is used, strict ordering could be achieved with the sticky hash range policy. In that case, the application needs to manage splitting the key range and ensuring that it gets fully covered by consumers. One possible solution is to use Kubernetes statefulset to run a specific number of replicas and have the application calculate the key range based on the current instance's pod ordinal index.
There's a similar challenge for strict ordered processing of partitioned topics. It seems that a lot of messaging use cases don't require strict ordered processing since there isn't much discussion about such application level solutions for strictly ordered message processing.
In Pulsar, failover subscription type supports ConsumerEventListener which could be used to detect when a specific consumer becomes active or deactivates in a specific instance. That might be useful in achieving strictly ordered message processing. However, from my own experiences, I think that the Kubernetes statefulset approach with a fixed strategy to choose partitions or key_shared hash ranges based on the pod ordinal index is very stable and easy to manage. The challenge becomes how to handle the case of scaling up or scaling down consumers.

There might be others in the Pulsar community that have come up with good solutions for specific ordered message processing use case.

I'd like to also mention that the Apache Pulsar Reactive client has great support for pipelining with high parallelism while processing messages in order. It uses Project Reactor's GroupedFlux to handle splitting the incoming messages to multiple streams of messages. Apache Pulsar Reactive client also integrates well with Spring Reactive and Spring Pulsar as well as any other Reactive Streams libraries. There are examples in https://github.com/lhotari/reactive-iot-backend-ApacheCon2021 and https://github.com/lhotari/reactive-pulsar-showcase of using Pulsar Reactive Client.

I hope this helps making progress in your Pulsar evaluation journey.

[niriven]

Thanks. This explains it well. It's a bit unnerving that this is the case with key_shared but I understand why .. because pulsar does not track individual keys, only assigned hash ranges for consumers.

We have a need to have one topic, and consumers process these messages in order for a given key. The same message may take 10 ms to process or 30 minutes depending upon internally defined customer logic.

I could work on splitting this into one topic per message per key to maintain ordering without blocking processing, but there could be millions of keys for us.

Ideally all keys produced to one key_shared subscription are still sent to (some consumer) for processing when re-balancing happens, even if it's the consumer that 'owns' the unacked messages -- then when acks catch up ranges are assigned appropriately to re-balance evenly.

I'll look at sticky hash, and possible use of allowOutOfOrderDelivery knowing that another consumer may receive a key for new messages when another consumer has not finished acking yet for the same key during consumer up / down operations.

[lhotari]

because pulsar does not track keys, only assigned hash ranges for consumers.

Yes, that's a current limitation. However one decision that I don't fully understand in the current implementation is to block delivery for all messages when any hash range is blocked. I guess the reason has been to minimize memory consumption. Perhaps a future improvement could be to allow tracking an individual hash value in the range 0-65535.

[lhotari]

Perhaps a future improvement could be to allow tracking an individual hash value in the range 0-65535.

@niriven This is being implemented as part of #23231. It makes improvements to how Key_Shared works internally and will make as much progress as possible on consumers and keys that aren't blocked. This will require reading ahead. Previously there weren't limits for reading ahead and this caused issue #23200. As part of fixing that issue there's now a limit and also tracking of individual hash values for blocking delivery only when there's an outstanding hash value. (Additional rules apply, which is for recently joined consumers, the updated logic described in PIP-282 and those changes were made earlier).

[lhotari]

There's yet another aspect of the individual hash tracking.

Currently, when new consumers join, all message delivery is blocked for the consumer until the inflight messages that were sent before the consumer joined have been processed. The reason for this behavior is to prevent out-of-order handling in cases since the hash assignments change in KeySharedMode.AUTO_SPLIT when consumers join and leave.

This solution would also benefit of individual hash tracking. PIP-282: Change definition of the recently joined consumers position will be delivered in Pulsar 4.0 and it contains important improvements. However, it seems that the solution isn't yet the most optimal design that would be possible and practical.

The updated design should also allow sending events to Pulsar clients which hash assignments change. There's an open issue #6555 about this. The common use case is to keep a cache and for cache invalidation, there's a need to know when a message key will no longer receive updates on a specific consumer when using the AUTO_SPLIT mode. A side note here is that for caching use cases, the AUTO_SPLIT is usually not applicable and there's already the STICKY mode where hash assignments can be controlled by the application.

The challenge with STICKY mode is that the application must make sure to cover the complete hash range. Implementing this on the application side is currently tedious and it would be useful to provide best practices and patterns of implementing this in Kubernetes. One possible solution is to use a Kubernetes StatefulSet and calculate the assigned sticky hash ranges based on the ordinal index of the pod. This way the hash range would always be handled by the same pod and there would be only a single instance. The challenge is when the StatefulSet is scaled up and down. The StatefulSet would have to be stopped completely before changing the StatefulSet size. The benefit of this solution is that it has worked already for years in Pulsar with Key_Shared subscriptions, but this pattern isn't well known.

[lhotari]

PIP-379: Key_Shared Draining Hashes for Improved Message Ordering scheduled for Pulsar 4.0 will address the problem. PIP-379 has been accepted by the Apache Pulsar project. The implementation PR is currently in review, #23352.

[lhotari]

PIP-379 will be included in Pulsar 4.0.0.