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