Fix hasMessageAvailable return true but can't read message

[315157973]

Motivation

I temporarily fixed this problem in PR #10190.
Now we have found a better way, this way can avoid the seek, then avoid trigger another reconnection.
Thank you @codelipenghui to troubleshoot this issue with me all night.

We have added a lot of log and found that this issue is caused by some race condition problems. Here is the first reason:

pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java

Lines 1808 to 1818 in f2d72c9

	cnx.sendRequestWithId(seek, requestId).thenRun(() -> {
	log.info("[{}][{}] Successfully reset subscription to {}", topic, subscription, seekBy);
	acknowledgmentsGroupingTracker.flushAndClean();
	seekMessageId = new BatchMessageIdImpl((MessageIdImpl) seekId);
	duringSeek.set(true);
	lastDequeuedMessageId = MessageId.earliest;
	clearIncomingMessages();
	seekFuture.complete(null);
	}).exceptionally(e -> {

Now we have an acknowledgmentsGroupingTracker to filter duplicate messages, and this Tracker will be cleaned up after seek.

However, it is possible that the connection is ready and Broker has pushed message, but acknowledgmentsGroupingTracker.flushAndClean(); has not been executed yet.

Finally hasMessageAvailableAsync returns true, but the message cannot be read because it is filtered by the acknowledgmentsGroupingTracker

Modifications

clean the tracker when connection was open

Verifying this change

[lhotari]

I wonder if #7796 and #7848 are anyhow related to this fix?

[315157973]

I wonder if #7796 and #7848 are anyhow related to this fix?

There are many reasons for this problem, this is just one of the scenarios, for example: there are messages in incomeQueue, but they cannot be consumed. there will be follow-up pr to fix this problem.
I've seen it, and this pr can't cover his scene
What can be confirmed now is that all the flaky tests in MultiTopicsReaderTest have been resolved

[wolfstudy]

Cool, LGTM +1