Fixed ordering issue in KeyShared dispatcher when adding consumer

[merlimat]

Note: this is based on top of #6791, #7104 and #7105. Once these are merged, I'll rebase here. For the sake of this review, check commit ad2e39b

Motivation

Fixes: #6554

Ordering is broken in KeyShared dispatcher if a new consumer c2 comes in and an existing consumer c1 goes out.

This is because messages with keys previously assigned to c1 may now route to c2.

The solution here is to have new consumers joining in a "paused" state.

For example, consider this sequence:

1. Subscriptions has c1 and c2 consumers
2. c3 joins. Some of the keys are now supposed to go to c3.
3. Instead of starting delivering to c3. We mark the current readPosition (let's call it rp0_c3) of the cursor for c3.
4. Any message that now hashes to c3 and that has messageId >= rp0_c3 will be deferred for later re-delivery
5. Any message that might get re-delivered (eg: originally sent to c1, but c1 has failed) to c3 and that has messageId < rp0_c3 will be sent to c3
6. When the markDelete position of the cursor will move past rp0_c3 the restriction on c3 will be lifted.

Essentially, c3 joins but can only receive old messages, until everything that was read before joining gets acked.

[codelipenghui]

@merlimat If I understand correctly all consumers can't get new messages if a new consumer joined right? The messages will add to the redelivery queue will lead to the next read entry operation replay the redelivery queue so that new messages can't keep going. This may cause an increase in consumption latency for all consumers. But, I think most of the existing consumers can continue to consume new messages because they do not overlap with the newly added consumer in any hash range.

So in the #6977, I added a buffer for the messages that need to wait for mark delete position processed. This can minimize the consumption latency jitter caused by the newly added consumer especially when consumers are hungry. This will be more useful for e2e latency-sensitive scenes. I want to listen to your thoughts, maybe my thoughts are more worrying.

Other than that, I have no objections to this PR. And except added a buffer, I think #6977 is the same, #6977 added a field in the consumer named fencePositionForKeyShared . Essentially, equivalent to recentlyJoinedConsumers in this PR.

[codelipenghui]

@merlimat I thought about it carefully. The buffer is still possible to break the message order dispatching of the same key. It is better not to add the buffer at present.

[merlimat]

The buffer is still possible to break the message order dispatching of the same key. It is better not to add the buffer at present.

Yes, that was my thought. To guarantee the ordering we need for these consumers to wait for that messages to be acked before start consuming.

On top of this I also have one more change with a setting for allowOutOfOrderDelivery. In some cases, an application might not need strict ordering, rather just the "stickiness" of keys to the same consumer. In this scenario, we can relax the ordering requirement and basically we skip the recentlyJoinedConsumers alltogether.

[codelipenghui]

On top of this I also have one more change with a setting for allowOutOfOrderDelivery. In some cases, an application might not need strict ordering, rather just the "stickiness" of keys to the same consumer. In this scenario, we can relax the ordering requirement and basically we skip the recentlyJoinedConsumers alltogether.

It sounds pretty good

[codelipenghui]

@merlimat Could you please help introduce the great approach at http://pulsar.apache.org/docs/en/concepts-messaging/#key_shared? This is very helpful for users to understand Key_Shared. I will merge this PR first.

On top of this I also have one more change with a setting for allowOutOfOrderDelivery. In some cases, an application might not need strict ordering, rather just the "stickiness" of keys to the same consumer. In this scenario, we can relax the ordering requirement and basically we skip the recentlyJoinedConsumers alltogether.

Looking forward to your new PR.