You are viewing a plain text version of this content. The canonical link for it is here.

Posted to commits@pulsar.apache.org by "niriven (via GitHub)" <gi...@apache.org> on 2024/06/14 18:35:49 UTC

[D] Behavior of Key_Shared with a partitioned topic? [pulsar]

GitHub user niriven created a discussion: Behavior of Key_Shared with a partitioned topic?

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 untrue 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 messages but messages for `key-0`
  * Consumer-1: `13:45:27.718 [pool-1-thread-1] INFO  c.t.q.p.c.PulsarSpecifiedKeySkipConsumer1 - 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 couple 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?

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

GitHub link: https://github.com/apache/pulsar/discussions/22912

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Key_Shared consumer scale behavior with a partitioned topic [pulsar]

Posted by "niriven (via GitHub)" <gi...@apache.org>.

GitHub user niriven edited a discussion: Key_Shared consumer scale behavior with a partitioned topic

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
* `.enableBatching(false)` set for both producers and consumers
* 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 false, it distributes as I'd expect (All consumers are working on all keys during scale up / down)

GitHub link: https://github.com/apache/pulsar/discussions/22912

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Key_Shared consumer scale behavior with a partitioned topic [pulsar]

Posted by "lhotari (via GitHub)" <gi...@apache.org>.

GitHub user lhotari added a comment to the discussion: Key_Shared consumer scale behavior with a partitioned topic

I'm currently working on a new design for Key_Shared AUTO_SPLIT mode which would cover the requirement of individual hash tracking. The goal is that a message will only get blocked when it is required for preserving ordering. #23231 improves the situation as well as the PIP-282 changes, but the goal isn't met. The remaining gap are the current "recently joined consumers" rules (which were significantly improved in PIP-282).

For STICKY mode, I have created https://github.com/apache/pulsar/pull/23275. Currently the "recently joined customer" rules are applied for STICKY mode which is causing unnecessary blocking.

There's an additional requirement that I'd like to cover, which is to have a way for the clients to detect when a specific consumer will no longer receive more messages for a particular message key after the assignments change in AUTO_SPLIT mode. There's a related issue https://github.com/apache/pulsar/issues/6555 about it, but it doesn't currently capture the actual requirement.

The use case is about being able to keep some client-side state for a particular message key and discard the state when the message key no longer will appear in the stream for a particular consumer. When the message key appears the first time, the client might build some entity and add it to a cache. When the message key is no longer assigned to the consumer, the client would discard it from the cache. No other consumer should be receiving a newer message with the same key until the client has acknowledged the last message with the message key.

To handle the ordering constraints and cover the requirement of notifying the clients about message key assignment, I'm working on a design that would cover this together with covering the goal of addressing the currently unoptimal "recently joined consumers" rules which could block message delivery when it's not necessary.

The exact details are still unknown, but I'm slowly making progress towards some concepts that would help work on the initial design.

One of the concepts that I'm thinking of is a "hash assignment generation," which is the assignment of consumers to the hashes at a particular point in time. When new consumers join and leave, a new generation would be formed. When the dispatcher sends messages to consumers, it could attach the hash of the message key and the hash assignment generation number to each message. In addition to this, each consumer would receive its own hash assignment information and the generation number of that assignment in a special marker message that is injected by the dispatcher into the messages that are delivered to the consumer.

A client library could use this information, for example, to flush client-side caches when it no longer continues to receive future messages for a particular key. The tracking of the key would be at the level of the hash for that key. When the consumer receives the hash information together with the message, there's no longer a need to recalculate the hash on the client side. This would allow also changing the hash algorithm on the broker side without a dependency in clients to know which hash algorithm was used.

I'll follow up with discussions on the dev mailing list which hopefully lead to a Pulsar Improvement Proposal (PIP) that could be addressed in the Pulsar 4.0 timeline.

GitHub link: https://github.com/apache/pulsar/discussions/22912#discussioncomment-10599466

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Key_Shared consumer scale behavior with a partitioned topic [pulsar]

Posted by "lhotari (via GitHub)" <gi...@apache.org>.

GitHub user lhotari added a comment to the discussion: Key_Shared consumer scale behavior with a partitioned topic

> 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.

GitHub link: https://github.com/apache/pulsar/discussions/22912#discussioncomment-9806707

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Key_Shared consumer scale behavior with a partitioned topic [pulsar]

Posted by "niriven (via GitHub)" <gi...@apache.org>.

GitHub user niriven edited a discussion: Key_Shared consumer scale behavior with a partitioned topic

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
* `.enableBatching(false)` set for producers and consumers
* 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 false, it distributes as I'd expect (All consumers are working on all keys during scale up / down)

GitHub link: https://github.com/apache/pulsar/discussions/22912

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Key_Shared consumer scale with a partitioned topic [pulsar]

Posted by "niriven (via GitHub)" <gi...@apache.org>.

GitHub user niriven edited a discussion: Key_Shared consumer scale with a partitioned topic

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 but `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 false, it distributes as I'd expect (All consumers are working on all keys during scale up / down)

GitHub link: https://github.com/apache/pulsar/discussions/22912

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Key_Shared consumer scale behavior with a partitioned topic [pulsar]

Posted by "niriven (via GitHub)" <gi...@apache.org>.

GitHub user niriven edited a comment on the discussion: Key_Shared consumer scale behavior with a partitioned topic

@lhotari Thanks. This explains it well. It's a bit unnerving that this is the case with key_shared but I understand why. 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 (and there are a small set of un-acked messages for a subset of the keys) -- but this doesn't seem to be the case.

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.

GitHub link: https://github.com/apache/pulsar/discussions/22912#discussioncomment-9808616

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Key_Shared consumer scale behavior with a partitioned topic [pulsar]

Posted by "niriven (via GitHub)" <gi...@apache.org>.

GitHub user niriven added a comment to the discussion: Key_Shared consumer scale behavior with a partitioned topic

@lhotari Thanks. This explains it well. It's a bit unnerving that this is the case with key_shared. We have a need to have one topic, and consumers process these messages in order. 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 (and there are a small set of un-acked messages for a subset of the keys) -- but this doesn't seem to be the case.

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.

GitHub link: https://github.com/apache/pulsar/discussions/22912#discussioncomment-9808616

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Key_Shared consumer scale behavior with a partitioned topic [pulsar]

Posted by "lhotari (via GitHub)" <gi...@apache.org>.

GitHub user lhotari added a comment to the discussion: Key_Shared consumer scale behavior with a partitioned topic

> Similar to an existing comment on the issue - When a service is no longer handling messages for a specific hash range I want to be able to evict the cache for that range.

@arhoads Makes sense. That seems to be a common use case which #6555 could help solving.

GitHub link: https://github.com/apache/pulsar/discussions/22912#discussioncomment-11549599

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Key_Shared consumer scale behavior with a partitioned topic [pulsar]

Posted by "niriven (via GitHub)" <gi...@apache.org>.

GitHub user niriven added a comment to the discussion: Key_Shared consumer scale behavior with a partitioned topic

@lhotari `.enableBatching(false)` is set to false for both the producer and consumer. I have updated that in my test steps above.

GitHub link: https://github.com/apache/pulsar/discussions/22912#discussioncomment-9793951

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Key_Shared consumer scale behavior with a partitioned topic [pulsar]

Posted by "niriven (via GitHub)" <gi...@apache.org>.

GitHub user niriven added a comment to the discussion: Key_Shared consumer scale behavior with a partitioned topic

@lhotari I did read about this but forgot to set it on the producer. I reran the test with `enableBatching(false)` on the producer.

I see similar results, just a bit different.
- When Consumer 1 initially comes online, all 10 keys are being received (while I am skipping, eg. not acknowledging `key-0` at all
- 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 still (`key-4`, `key-6`, `key-8`)

GitHub link: https://github.com/apache/pulsar/discussions/22912#discussioncomment-9794422

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Key_Shared consumer scale behavior with a partitioned topic [pulsar]

Posted by "niriven (via GitHub)" <gi...@apache.org>.

GitHub user niriven edited a discussion: Key_Shared consumer scale behavior with a partitioned topic

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-poc.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-poc.zip](https://github.com/user-attachments/files/15876120/durable-queue-poc.zip)


GitHub link: https://github.com/apache/pulsar/discussions/22912

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Key_Shared consumer scale behavior with a partitioned topic [pulsar]

Posted by "niriven (via GitHub)" <gi...@apache.org>.

GitHub user niriven edited a comment on the discussion: Key_Shared consumer scale behavior with a partitioned topic

@lhotari 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 keys, only assigned hash ranges. 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 (and there are a small set of un-acked messages for a subset of the keys) -- but this doesn't seem to be the case.

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.

GitHub link: https://github.com/apache/pulsar/discussions/22912#discussioncomment-9808616

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Key_Shared consumer scale behavior with a partitioned topic [pulsar]

Posted by "niriven (via GitHub)" <gi...@apache.org>.

GitHub user niriven edited a comment on the discussion: Key_Shared consumer scale behavior with a partitioned topic

@lhotari Thank you. I forgot to set it on the producer. I reran the test with `enableBatching(false)` on the producer.

I see similar results, just a bit different.
- 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 still (`key-4`, `key-6`, `key-8`)

GitHub link: https://github.com/apache/pulsar/discussions/22912#discussioncomment-9794422

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Key_Shared consumer scale behavior with a partitioned topic [pulsar]

Posted by "niriven (via GitHub)" <gi...@apache.org>.

GitHub user niriven edited a comment on the discussion: Key_Shared consumer scale behavior with a partitioned topic

@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`)

GitHub link: https://github.com/apache/pulsar/discussions/22912#discussioncomment-9794422

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Key_Shared consumer scale behavior with a partitioned topic [pulsar]

Posted by "niriven (via GitHub)" <gi...@apache.org>.

GitHub user niriven edited a discussion: Key_Shared consumer scale behavior with a partitioned topic

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)

GitHub link: https://github.com/apache/pulsar/discussions/22912

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Behavior of Key_Shared with a partitioned topic? [pulsar]

Posted by "niriven (via GitHub)" <gi...@apache.org>.

GitHub user niriven edited a discussion: Behavior of Key_Shared with a partitioned topic?

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 but `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 false, it distributes as I'd expect (All consumers are working on all keys during scale up / down)

GitHub link: https://github.com/apache/pulsar/discussions/22912

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Key_Shared consumer scale behavior with a partitioned topic [pulsar]

Posted by "lhotari (via GitHub)" <gi...@apache.org>.

GitHub user lhotari added a comment to the discussion: Key_Shared consumer scale behavior with a partitioned topic

>  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.

GitHub link: https://github.com/apache/pulsar/discussions/22912#discussioncomment-9808993

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Key_Shared consumer scale behavior with a partitioned topic [pulsar]

Posted by "niriven (via GitHub)" <gi...@apache.org>.

GitHub user niriven edited a discussion: Key_Shared consumer scale behavior with a partitioned topic

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 false, it distributes as I'd expect (All consumers are working on all keys during scale up / down)

GitHub link: https://github.com/apache/pulsar/discussions/22912

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Key_Shared consumer scale behavior with a partitioned topic [pulsar]

Posted by "lhotari (via GitHub)" <gi...@apache.org>.

GitHub user lhotari added a comment to the discussion: Key_Shared consumer scale behavior with a partitioned topic

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

This is addressed in PIP-379, summary available in the mailing list thread, https://lists.apache.org/thread/l5zjq0fb2dscys3rsn6kfl7505tbndlx .

GitHub link: https://github.com/apache/pulsar/discussions/22912#discussioncomment-10650631

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Key_Shared consumer scale behavior with a partitioned topic [pulsar]

Posted by "niriven (via GitHub)" <gi...@apache.org>.

GitHub user niriven added a comment to the discussion: Key_Shared consumer scale behavior with a partitioned topic

@lhotari 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.

GitHub link: https://github.com/apache/pulsar/discussions/22912#discussioncomment-9808927

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Behavior of Key_Shared with a partitioned topic? [pulsar]

Posted by "niriven (via GitHub)" <gi...@apache.org>.

GitHub user niriven edited a discussion: Behavior of Key_Shared with a partitioned topic?

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 messages but messages for `key-0`
  * 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 false, it distributes as I'd expect (All consumers are working on all keys during scale up / down)

GitHub link: https://github.com/apache/pulsar/discussions/22912

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Behavior of Key_Shared with a partitioned topic during consumer scale [pulsar]

Posted by "niriven (via GitHub)" <gi...@apache.org>.

GitHub user niriven edited a discussion: Behavior of Key_Shared with a partitioned topic during consumer scale

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 but `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 false, it distributes as I'd expect (All consumers are working on all keys during scale up / down)

GitHub link: https://github.com/apache/pulsar/discussions/22912

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Key_Shared consumer scale behavior with a partitioned topic [pulsar]

Posted by "lhotari (via GitHub)" <gi...@apache.org>.

GitHub user lhotari edited a comment on the discussion: Key_Shared consumer scale behavior with a partitioned topic

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](https://pulsar.apache.org/docs/3.3.x/concepts-messaging/#batching-for-key-shared-subscriptions)).
Are you handling this in your test configuration?



GitHub link: https://github.com/apache/pulsar/discussions/22912#discussioncomment-9793388

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Key_Shared consumer scale behavior with a partitioned topic [pulsar]

Posted by "lhotari (via GitHub)" <gi...@apache.org>.

GitHub user lhotari added a comment to the discussion: Key_Shared consumer scale behavior with a partitioned topic

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:

```shell
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](https://gist.github.com/lhotari/694792fc91c216c8626af27fe0b9b0ef#file-stats-json)
* [stats_internal.json](https://gist.github.com/lhotari/694792fc91c216c8626af27fe0b9b0ef#file-stats_internal-json)

In stats_internal, you have the [individuallyDeletedMessages](https://gist.github.com/lhotari/694792fc91c216c8626af27fe0b9b0ef#file-stats_internal-json-L34) information which shows the message ranges after [markDeletePosition](https://gist.github.com/lhotari/694792fc91c216c8626af27fe0b9b0ef#file-stats_internal-json-L27) that have already been acknowledged. When the range is continuous, the markDeletePosition will move forward.

In internal stats, the [numberOfEntriesSinceFirstNotAckedMessage](https://gist.github.com/lhotari/694792fc91c216c8626af27fe0b9b0ef#file-stats_internal-json-L38) 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](https://gist.github.com/lhotari/694792fc91c216c8626af27fe0b9b0ef#file-stats-json-L41). [unackedMessages information is available in the stats at the consumer level](https://gist.github.com/lhotari/694792fc91c216c8626af27fe0b9b0ef#file-stats-json-L65) 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](https://github.com/apache/pulsar/blob/master/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageRedeliveryController.java). 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:
https://github.com/apache/pulsar/blob/411f6973e85b0a6213e992386e1704f93d0aae42/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java#L344-L349
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](https://pulsar.apache.org/docs/next/client-libraries-producers/#publish-messages-to-partitioned-topics) explains how publishing works.

The [key_shared documentation](https://pulsar.apache.org/docs/next/concepts-messaging/#key_shared) 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](https://pulsar.apache.org/docs/next/concepts-messaging/#key_shared). 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](https://kubernetes.io/docs/concepts/workloads/controllers/statefulset/#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](https://pulsar.apache.org/api/client/3.1.x/org/apache/pulsar/client/api/ConsumerEventListener.html) 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.

GitHub link: https://github.com/apache/pulsar/discussions/22912#discussioncomment-9808314

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Key_Shared consumer scale behavior with a partitioned topic [pulsar]

Posted by "niriven (via GitHub)" <gi...@apache.org>.

GitHub user niriven edited a comment on the discussion: Key_Shared consumer scale behavior with a partitioned topic

@lhotari 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 (and there are a small set of un-acked messages for a subset of the keys) -- but this doesn't seem to be the case.

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.

GitHub link: https://github.com/apache/pulsar/discussions/22912#discussioncomment-9808616

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Key_Shared consumer scale behavior with a partitioned topic [pulsar]

Posted by "niriven (via GitHub)" <gi...@apache.org>.

GitHub user niriven edited a discussion: Key_Shared consumer scale behavior with a partitioned topic

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](https://github.com/user-attachments/files/15877712/durable-queue-poc2.zip)


GitHub link: https://github.com/apache/pulsar/discussions/22912

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Key_Shared consumer scale behavior with a partitioned topic [pulsar]

Posted by "arhoads (via GitHub)" <gi...@apache.org>.

GitHub user arhoads added a comment to the discussion: Key_Shared consumer scale behavior with a partitioned topic

Similar to an existing comment on the issue - When a service is no longer handling messages for a specific hash range I want to be able to evict the cache for that range. 

GitHub link: https://github.com/apache/pulsar/discussions/22912#discussioncomment-11527739

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Key_Shared consumer scale behavior with a partitioned topic [pulsar]

Posted by "arhoads (via GitHub)" <gi...@apache.org>.

GitHub user arhoads added a comment to the discussion: Key_Shared consumer scale behavior with a partitioned topic

Are there still plans to add a hash range change event for consumers?

GitHub link: https://github.com/apache/pulsar/discussions/22912#discussioncomment-11466505

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Key_Shared consumer scale behavior with a partitioned topic [pulsar]

Posted by "lhotari (via GitHub)" <gi...@apache.org>.

GitHub user lhotari added a comment to the discussion: Key_Shared consumer scale behavior with a partitioned topic

> Are there still plans to add a hash range change event for consumers?

@arhoads Yes, but there isn't yet a schedule. Now, after PIP-379, it will be easier to implement https://github.com/apache/pulsar/issues/6555 . 

Do you have a specific use case for the change events? What problem would it solve in your case?

GitHub link: https://github.com/apache/pulsar/discussions/22912#discussioncomment-11469915

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Behavior of Key_Shared with a partitioned topic? [pulsar]

Posted by "niriven (via GitHub)" <gi...@apache.org>.

GitHub user niriven edited a discussion: Behavior of Key_Shared with a partitioned topic?

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 untrue 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 messages but messages for `key-0`
  * 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 false, it distributes as I'd expect (All consumers are working on all keys during scale up / down)

GitHub link: https://github.com/apache/pulsar/discussions/22912

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Key_Shared consumer scale behavior with a partitioned topic [pulsar]

Posted by "niriven (via GitHub)" <gi...@apache.org>.

GitHub user niriven edited a discussion: Key_Shared consumer scale behavior with a partitioned topic

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 but `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 false, it distributes as I'd expect (All consumers are working on all keys during scale up / down)

GitHub link: https://github.com/apache/pulsar/discussions/22912

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Behavior of Key_Shared with a partitioned topic? [pulsar]

Posted by "niriven (via GitHub)" <gi...@apache.org>.

GitHub user niriven edited a discussion: Behavior of Key_Shared with a partitioned topic?

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 but messages `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 false, it distributes as I'd expect (All consumers are working on all keys during scale up / down)

GitHub link: https://github.com/apache/pulsar/discussions/22912

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Key_Shared consumer scale behavior with a partitioned topic [pulsar]

Posted by "lhotari (via GitHub)" <gi...@apache.org>.

GitHub user lhotari added a comment to the discussion: Key_Shared consumer scale behavior with a partitioned topic

[PIP-379: Key_Shared Draining Hashes for Improved Message Ordering](https://github.com/apache/pulsar/blob/master/pip/pip-379.md) 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.

GitHub link: https://github.com/apache/pulsar/discussions/22912#discussioncomment-10848746

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Key_Shared consumer scale behavior with a partitioned topic [pulsar]

Posted by "lhotari (via GitHub)" <gi...@apache.org>.

GitHub user lhotari added a comment to the discussion: Key_Shared consumer scale behavior with a partitioned topic

> 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).

GitHub link: https://github.com/apache/pulsar/discussions/22912#discussioncomment-10568733

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Key_Shared consumer scale behavior with a partitioned topic [pulsar]

Posted by "lhotari (via GitHub)" <gi...@apache.org>.

GitHub user lhotari edited a comment on the discussion: Key_Shared consumer scale behavior with a partitioned topic

For STICKY mode, I have created https://github.com/apache/pulsar/pull/23275. Currently the "recently joined customer" rules are applied for STICKY mode which is causing unnecessary blocking.

For AUTO_SPLIT mode, there's an additional requirement that I'd like to cover, which is to have a way for the clients to detect when a specific consumer will no longer receive more messages for a particular message key after the assignments change in AUTO_SPLIT mode. There's a related issue https://github.com/apache/pulsar/issues/6555 about it, but it doesn't currently capture the actual requirement.

The exact details are still unknown, but I'm slowly making progress towards some concepts that would help work on the initial design.

I'll follow up with discussions on the dev mailing list which hopefully lead to a Pulsar Improvement Proposal (PIP) that could be addressed in the Pulsar 4.0 timeline.

GitHub link: https://github.com/apache/pulsar/discussions/22912#discussioncomment-10599466

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Key_Shared consumer scale behavior with a partitioned topic [pulsar]

Posted by "niriven (via GitHub)" <gi...@apache.org>.

GitHub user niriven edited a comment on the discussion: Key_Shared consumer scale behavior with a partitioned topic

@lhotari 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 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 (and there are a small set of un-acked messages for a subset of the keys) -- but this doesn't seem to be the case.

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.

GitHub link: https://github.com/apache/pulsar/discussions/22912#discussioncomment-9808616

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Key_Shared consumer scale behavior with a partitioned topic [pulsar]

Posted by "niriven (via GitHub)" <gi...@apache.org>.

GitHub user niriven edited a discussion: Key_Shared consumer scale behavior with a partitioned topic

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 but `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 false, it distributes as I'd expect (All consumers are working on all keys during scale up / down)

GitHub link: https://github.com/apache/pulsar/discussions/22912

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Key_Shared consumer scale behavior with a partitioned topic [pulsar]

Posted by "niriven (via GitHub)" <gi...@apache.org>.

GitHub user niriven edited a comment on the discussion: Key_Shared consumer scale behavior with a partitioned topic

@lhotari Thank you. I forgot to set it on the producer. I reran the test with `enableBatching(false)` on the producer.

I see similar results, just a bit different.
- 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`)

GitHub link: https://github.com/apache/pulsar/discussions/22912#discussioncomment-9794422

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Key_Shared consumer scale behavior with a partitioned topic [pulsar]

Posted by "niriven (via GitHub)" <gi...@apache.org>.

GitHub user niriven edited a comment on the discussion: Key_Shared consumer scale behavior with a partitioned topic

@lhotari 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 things catch 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.

GitHub link: https://github.com/apache/pulsar/discussions/22912#discussioncomment-9808616

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Key_Shared consumer scale behavior with a partitioned topic [pulsar]

Posted by "niriven (via GitHub)" <gi...@apache.org>.

GitHub user niriven edited a comment on the discussion: Key_Shared consumer scale behavior with a partitioned topic

@lhotari 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 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 (and there are a small set of un-acked messages for a subset of the keys) -- but this doesn't seem to be the case.

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.

GitHub link: https://github.com/apache/pulsar/discussions/22912#discussioncomment-9808616

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Key_Shared consumer scale behavior with a partitioned topic [pulsar]

Posted by "niriven (via GitHub)" <gi...@apache.org>.

GitHub user niriven edited a comment on the discussion: Key_Shared consumer scale behavior with a partitioned topic

@lhotari Thanks. This explains it well. It's a bit unnerving that this is the case with key_shared but I understand why. We have a need to have one topic, and consumers process these messages in order. 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 (and there are a small set of un-acked messages for a subset of the keys) -- but this doesn't seem to be the case.

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.

GitHub link: https://github.com/apache/pulsar/discussions/22912#discussioncomment-9808616

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Key_Shared consumer scale behavior with a partitioned topic [pulsar]

Posted by "niriven (via GitHub)" <gi...@apache.org>.

GitHub user niriven edited a discussion: Key_Shared consumer scale behavior with a partitioned topic

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 false, it distributes as I'd expect (All consumers are working on all keys during scale up / down)

GitHub link: https://github.com/apache/pulsar/discussions/22912

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Key_Shared consumer scale behavior with a partitioned topic [pulsar]

Posted by "lhotari (via GitHub)" <gi...@apache.org>.

GitHub user lhotari added a comment to the discussion: Key_Shared consumer scale behavior with a partitioned topic

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.
Are you handling this in your test configuration?



GitHub link: https://github.com/apache/pulsar/discussions/22912#discussioncomment-9793388

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Key_Shared consumer scale behavior with a partitioned topic [pulsar]

Posted by "lhotari (via GitHub)" <gi...@apache.org>.

GitHub user lhotari added a comment to the discussion: Key_Shared consumer scale behavior with a partitioned topic

PIP-379 will be included in Pulsar 4.0.0.

GitHub link: https://github.com/apache/pulsar/discussions/22912#discussioncomment-10962538

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Key_Shared consumer scale behavior with a partitioned topic [pulsar]

Posted by "lhotari (via GitHub)" <gi...@apache.org>.

GitHub user lhotari added a comment to the discussion: Key_Shared consumer scale behavior with a partitioned topic

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](https://github.com/apache/pulsar/blob/master/pip/pip-282.md) 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 https://github.com/apache/pulsar/issues/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.

GitHub link: https://github.com/apache/pulsar/discussions/22912#discussioncomment-10589453

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Key_Shared consumer scale behavior with a partitioned topic [pulsar]

Posted by "niriven (via GitHub)" <gi...@apache.org>.

GitHub user niriven deleted a comment on the discussion: Key_Shared consumer scale behavior with a partitioned topic

@lhotari `.enableBatching(false)` is set to false for both the producer and consumer. I have updated that in my test steps above.

GitHub link: https://github.com/apache/pulsar/discussions/22912#discussioncomment-9793951

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Key_Shared consumer scale behavior with a partitioned topic [pulsar]

Posted by "niriven (via GitHub)" <gi...@apache.org>.

GitHub user niriven edited a comment on the discussion: Key_Shared consumer scale behavior with a partitioned topic

@lhotari Thank you. I forgot to set it on the producer. I reran the test with `enableBatching(false)` on the producer.

I see similar results, just a bit different.
- When Consumer 1 initially comes online, all 10 keys are being received (while I am skipping, eg. not acknowledging `key-0` at all
- 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 still (`key-4`, `key-6`, `key-8`)

GitHub link: https://github.com/apache/pulsar/discussions/22912#discussioncomment-9794422

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Key_Shared consumer scale behavior with a partitioned topic [pulsar]

Posted by "niriven (via GitHub)" <gi...@apache.org>.

GitHub user niriven edited a comment on the discussion: Key_Shared consumer scale behavior with a partitioned topic

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.

GitHub link: https://github.com/apache/pulsar/discussions/22912#discussioncomment-9808927

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Key_Shared consumer scale behavior with a partitioned topic [pulsar]

Posted by "niriven (via GitHub)" <gi...@apache.org>.

GitHub user niriven deleted a comment on the discussion: Key_Shared consumer scale behavior with a partitioned topic

@lhotari 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.

GitHub link: https://github.com/apache/pulsar/discussions/22912#discussioncomment-9808616

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Key_Shared consumer scale behavior with a partitioned topic [pulsar]

Posted by "niriven (via GitHub)" <gi...@apache.org>.

GitHub user niriven edited a comment on the discussion: Key_Shared consumer scale behavior with a partitioned topic

@lhotari 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.

GitHub link: https://github.com/apache/pulsar/discussions/22912#discussioncomment-9808616

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Key_Shared consumer scale behavior with a partitioned topic [pulsar]

Posted by "niriven (via GitHub)" <gi...@apache.org>.

GitHub user niriven edited a comment on the discussion: Key_Shared consumer scale behavior with a partitioned topic

@lhotari 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.

GitHub link: https://github.com/apache/pulsar/discussions/22912#discussioncomment-9808616

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Key_Shared consumer scale behavior with a partitioned topic [pulsar]

Posted by "niriven (via GitHub)" <gi...@apache.org>.

GitHub user niriven edited a comment on the discussion: Key_Shared consumer scale behavior with a partitioned topic

@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.

I see similar results, just a bit different.
- 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`)

GitHub link: https://github.com/apache/pulsar/discussions/22912#discussioncomment-9794422

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org

Re: [D] Behavior of Key_Shared with a partitioned topic? [pulsar]

Posted by "niriven (via GitHub)" <gi...@apache.org>.

GitHub user niriven edited a discussion: Behavior of Key_Shared with a partitioned topic?

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 untrue 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 messages but messages for `key-0`
  * Consumer-1: `13:45:27.718 [pool-1-thread-1] INFO  c.t.q.p.c.PulsarSpecifiedKeySkipConsumer1 - 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 couple 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?

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

GitHub link: https://github.com/apache/pulsar/discussions/22912

----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org