[fix][client]Duplicate messages when use MultiTopicsConsumerImpl

[poorbarcode]

Fixes

• When using the Multi Topics Consumer, the message will be consumed repeatedly
• Also fix these flaky test:
  • [fix][test] Fix flaky test TopicReaderTest.testMultiReaderIsAbleToSeekWithTimeOnMiddleOfTopic #17147
  • Flaky-test: TopicReaderTest.testMultiReaderIsAbleToSeekWithTimeOnMiddleOfTopic #17407

Motivation

There has a race condition between add messages to incoming queue and clean incoming queue.

---

When we call MultiTopicsConsumerImpl.seek, we expect the process executed like this:

• clear incoming queue
• each consumer executes seek
  • consumer_1 seek
  • consumer_2 seek
• wait for all consumers seek finished
• receive messages

---

But the real process might be executed like this:

• clear incoming queue
• (High light)receiveMessageFromConsumer is being executed at this time, many messages being added to the incoming queue
• each consumer executes seek
  • consumer_1 seek
  • consumer_2 seek
• (High light)receiveMessageFromConsumer is being executed at step 2, and many messages are being added to the incoming queue at this step
• wait for all consumers seek finished
• receive messages

then the same message will be consumed twice:

• add message(3:3) to the incoming queue
• seek to 3:1
• add message[ (3:1),(3:2),(3:3) ] to the incoming queue
• (High light) incoming queue has 4 messages [ (3:3), (3:1),(3:2),(3:3) ]

---

How to reproduce?

execute TopicReaderTest.testMultiReaderIsAbleToSeekWithTimeOnMiddleOfTopic 100 ～400 times

---

Modifications

Make seek to two parts action:

part-1:

• seek start: call pause
• if the status is pause, all action receiveMessageFromConsumer will not execute immediately, just into the pending queue

part-2:

• wait all consumers seek finished, call resume
• execute all actions in the pending queue

Documentation

• doc-required

• doc-not-needed

• doc

• doc-complete

[poorbarcode]

This PR should merge into these branches (because this test was appended in branch-2.11):

• branch-2.8
• branch-2.9
• branch-2.10
• branch-2.11
• master

Note: new features have been added in branch-2.8, so if we want to cherry-pick to branch-2.7, there will be conflicts.

[poorbarcode]

/pulsarbot rerun-failure-checks

[codelipenghui]

@poorbarcode Could you please provide more context about how the race condition will happen?

[poorbarcode]

Hi @codelipenghui

Already added the description. I'm sorry I didn't finish TODO in time

[MarvinCai]

do we need to wait for completion of these ongoing receives as the messages will be discarded anyway, can we just cancel them?

[poorbarcode]

Good suggestion, but we don't have the operability of cancel, and the waiting-event only happens on this scenario: call seek multi times quickly, along with read, this is very rare.

[codelipenghui]

@poorbarcode Thanks. I understand the problem for now.

There is ML discussion https://lists.apache.org/thread/97o9t4ltkds5pfq41l9xbbd31t41qm8w
I think we'd better find a general solution for the message duplication

And this one https://lists.apache.org/thread/gnqwxo7w6n6g72ochvgpgv4s6r8mnwb7 is also a duplicated message
discussion.

These are not exactly the same problem, but tight related.

For this PR. we will introduce many queue operations to avoid duplicated messages.
This seems to be a bit expensive.

[poorbarcode]

PIP-194 will fix this problem

• PIP 194 : Pulsar client: seek command add epoch #16757