[fix][broker] Avoid messages being repeatedly replayed with SHARED subscriptions (streaming dispatcher)

[nicoloboschi]

Motivation

There's a test failing constantly in the flaky suite:
#16795

While investigating, I found out a case when it's possible that the dispatcher replays twice the same message to the consumer. This may happens because the message's replay is async and it's done while reading entries.

I believe that it may be related to recent changes to the dispatcher:

• [enh] Broker - Shared subscription: run filters in a separate (per-subscription) thread (dispatcherDispatchMessagesInSubscriptionThread) #16603
• Issue 16802: fix Repeated messages of shared dispatcher #16812
• [fix][broker]Prevent StackOverFlowException in SHARED subscription #16968
• [fix][broker] Streaming dispatcher stuck after reading the first entry with SHARED subscriptions #17143

Modifications

• Added the same guard inside readMoreEntries for the streaming dispatcher

The proof that this fix works is that now the test SimpleProducerConsumerTest.testSharedSamePriorityConsumer doesn't fail anymore.

Fix #16795

• doc-not-needed

[eolivelli]

good catch !

we need more eyes on this patch

@Technoboy- @codelipenghui @hangc0276 @merlimat @lhotari PTAL

[nicoloboschi]

@eolivelli thanks for the review. The PersistentStickyKeyDispatcherMultipleConsumers had the same issue. I fixed it and added a test

[mattisonchao]

I think we have to know the root cause.

[codelipenghui]

I think we have to know the root cause.

+1, we should handle this PR with extra care. It will filter out some messages,
This may cause the message to not be delivered correctly to the client under some conditions.

[mattisonchao]

Hi, @nicoloboschi

The proof that this fix works is that now the test SimpleProducerConsumerTest.testSharedSamePriorityConsumer doesn't fail anymore.

I think it should be SimpleProducerConsumerTestStreamingDispatcherTest.testSharedSamePriorityConsumer

By the way, I want to collect some tests that need to move out of the flaky-test group... Because in the current time we may lose some test coverage.
See #17145

[mattisonchao]

After the test, I think some race conditions in multi-thread should cause it.
I try to disable this configuration:
dispatcherDispatchMessagesInSubscriptionThread run the test 30 times, everything got fine.

Controlled experiment:

[codelipenghui]

Thanks @mattisonchao

I think we should point out the race condition or revert dispatcherDispatchMessagesInSubscriptionThread in the 2.11.0 release. It will be a potential risk, and there may be other issues just haven't break the test.

[nicoloboschi]

Thanks for the reviews @codelipenghui and @mattisonchao.
I did additional testing and I still think this solution is correct.

Basically the case I'm seeing is the following:

1. After - normally - reading an entry, the dispatcher calls sendMessagesToConsumers.
2. The consumers are busy (actually the test properly verify the busy consumers) and so no permits are available for any consumer. The message is added to the redeliveryMessages structure.
   3.a In the next read cycle (readMoreEntries) the replay read is triggered asynchronously.
   3.b At the same time a consumer is ready to receive a new message and then consumerFlow is triggered. And, again, readMoreEntries.
3. Since the cursor read and replay read points the same position, the entry is read twice. (That could be another unwanted issue and may impact performance)
4. After the read is complete, the send is triggered. The send method doesn't check if in case of replay the message is still on the redeliveryMessages structure. So the message is sent twice (sequentially)

After the recent work (see mentioned pull in the description) the methods sendMessagesToConsumers and readMoreEntries are no more executed in the same lock and that's why this situation wouldn't have happened before.

The point 4. could be a problem and a new behaviour that will be introduced in 2.11. With this fix we prefer consistency (no duplicated sent to the consumers) over duplicated-reading. We read the same entry twice and perhaps we could optimize this edge-case in another pull. IMO still think this whole work of using different threads is worth.

[eolivelli]

+1

[mattisonchao]

Hi, @nicoloboschi

3.a In the next read cycle (readMoreEntries) the replay read is triggered asynchronously.
3.b At the same time a consumer is ready to receive a new message and then consumerFlow is triggered. And, again,

I'm still wondering why invoking readMoreEntires concurrently will got race conditions. I find we may lose the important state sendInProgress when invoking PersistentStreamingDispatcherMultipleConsumers#readMoreEntires method, After add this state control. the test never failed.

Controlled experiment:

[nicoloboschi]

@mattisonchao nice catch!! you're right, that is the missing check
That also explains why only the streaming test fails.

Thanks, I've updated the pull with your suggestion

[mattisonchao]

I think we have to check the relative logic, to avoid another problem. Because I think this kind of logic is very important.

[mattisonchao]

Could we move the test out of the flaky group to avoid regression?

[eolivelli]

I am sorry.
I misread the PR.
I thought we were in another class.

+1
Nice team work @nicoloboschi @mattisonchao

[eolivelli]

@Technoboy- this change must go into 2.11 as well