Hi,
I am exploring the Key_Shared feature of Apache Pulsar using the Pulsar
2.10.1 client.
*UseCase Perspective:*
I am simulating an everyday use case in Kubernetes-based deployments.
when we deploy new code to the pod's consumers restart, the first consumer
that starts is picking all the backlog, and other consumers are idle until
new messages are produced to the topic.
*Below is the consumer code configuration.*
consumer = initClient()
.newConsumer(Schema.STRING)
.topic(topicName)
.subscriptionType(SubscriptionType.Key_Shared)
.subscriptionName("test123")
.keySharedPolicy(KeySharedPolicy.autoSplitHashRange()
.setAllowOutOfOrderDelivery(true))
.messageListener(this::readMessage)
.consumerName(UUID.randomUUID().toString())
.subscribe();
I have created a group of 3 consumers to consume data and followed the
below steps.
1. Produce messages with randomly generated key and value pair
2. Now, The keys are distributed to all my consumers while consuming. And
all the messages are acknowledged. This is working great, as expected
3. Now, stop all the consumers and create a backlog. About 1000 random key
values pairs
4. Now, start all your consumers.
5. The issue is entire backlog is mapped to first consumer, and all the
other consumers are just passed, not receiving any event
6. When I produce new messages. They are hashed, and all 3 consumers are
receiving new messages
*My conclusions on a high level:*
1. Messages that are already hashed to the first consumer are not
distributed to the newly added consumers. Even if the first consumer picks
them up slowly
2. But when this first consumer is taken down, both the other consumers are
receiving backlog events
3. when I used setAllowOutOfOrderDelivery(true) new messages are sent to
newly added consumers. But the backlog is still stuck to just one consumer.
4. Data produced after the consumers are added is hashed properly, but the
older data is just stuck to one consumer.
Also, I have not seen this case in test cases of
KeySharedSubscriptionTest, regarding the distribution percentage of already
produced events with setAllowOutOfOrderDelivery(true).
Thank you,
Tarun.
