[consumer] Revert "Remove consumer unnecessary locks (#9261)"

[linlinnn]

Fixes #10222
Fixes #10173

Motivation

fix order guarantee for MultiTopicsConsumerImpl

The cause of out of order:

the origin order of message: 1, 2, 3, 4

// ConsumerImpl receive message
if (peekPendingReceive() != null) { //code link 1
    notifyPendingReceivedCallback(message, null); // code link 2
} else if (enqueueMessageAndCheckBatchReceive(message) /** code link 3  **/&& hasPendingBatchReceive()) {
    notifyPendingBatchReceivedCallBack();
}

// ConsumerImpl#internalReceiveAsync
Message<T> message = incomingMessages.poll(); // code link 4 
if (message == null) { 
    pendingReceives.add(result); // code link 5
    cancellationHandler.setCancelAction(() -> pendingReceives.remove(result));
} else {
    messageProcessed(message);
    result.complete(beforeConsume(message));
}

receive message 3
Thread pulsar-client-io ConsumerImpl peek pending receive is null (code link 1)
Thread pulsar-client-internal ConsumerImpl poll message is null (code link 4)
Thread pulsar-client-internal ConsumerImpl pending receive for message 3 (code link 5)
Thread pulsar-client-io ConsumerImpl offer message 3 to incomingMessages (code link 3)
receive message 4
Thread pulsar-client-io ConsumerImpl send message 4 notify above pending receive (code link 2)
Thread pulsar-client-internal MultiTopicsConsumerImpl accept message 4
Thread pulsar-client-internal MultiTopicsConsumerImpl accept message 3

Modification

Revert "Remove consumer unnecessary locks (#9261)"

[linlinnn]

@MarvinCai @codelipenghui @merlimat Please take a look.

[codelipenghui]

@linlinnn I have some doubts here, why we need to guarantee message order across partitions(The MultiTopicsConsumerImpl is consuming messages from multiple partitions)?

[linlinnn]

@linlinnn I have some doubts here, why we need to guarantee message order across partitions

@codelipenghui
Yes, we don't need to guarantee message order across partitions, but we should guarantee message order from the same partition, The MultiTopicsConsumerImpl iterates ConsumerImpl(consuming messages from single partition) to consume messages from multiple topics and multiple partitions. In this issue, The MultiTopicsConsumerImpl broken the order of the message from the same partition

newConsumers.forEach(consumer -> {
                // each ConsumerImpl consume messages from single partition
                consumer.increaseAvailablePermits(consumer.getConnectionHandler().cnx(), conf.getReceiverQueueSize());
                receiveMessageFromConsumer(consumer);
            });

[MarvinCai]

I think we should guarantee order by process incoming message from single thread, instead of adding lock which can impact performance due to context switch and wait.

[linlinnn]

I think we should guarantee order by process incoming message from single thread, instead of adding lock which can impact performance due to context switch and wait.

@MarvinCai
Emm. I agree your consideration. But the concurrent logic notify and offer seems conflict with single thread without lock.

[linlinnn]

/pulsarbot run-failure-checks

[315157973]

This is indeed a problem

        //1
       Message<T> message = incomingMessages.poll();
        if (message == null) {
         //2
            pendingReceives.add(result);
            cancellationHandler.setCancelAction(() -> pendingReceives.remove(result));
        } else {
         //3
            messageProcessed(message);
            result.complete(beforeConsume(message));
        }

        // 4
        if (peekPendingReceive() != null) {
            notifyPendingReceivedCallback(message, null);
        }
       ...
       // Message enters incomeQueue

When the execution order is 1、4、2、1、3、4, the order of 3 and 4 cannot be guaranteed

[315157973]

Emm...We have orderExecutor CC @hangc0276

[linlinnn]

/pulsarbot run-failure-checks

[MarvinCai]

I actually don't see the point of the locking we try to add here. The poll from receiver queue or notify pending request is not the root cause here. The BlockingQueue is itself threadsafe. And no matter which thread does the notification, it'll alway be thread from "pulsar-external-listener" to process pending request.
The root cause is when message is retrieved in the line, the receiveAsync can be completed from 2 different thread in different situation, if there are messages in receiver queue, it'll be completed from "pulsar-client-internal" thread, if it become pending request, it'll later be completed from "pulsar-external-listener" thread. Even if order of message arrive the underlying single ConsumerImpl is correct, the MultiTopicsConsumerImpl can see out of order message cause 2 thread are processing message independently.
This only happen with multitopic consumer in case of small message volume without proper batching like in out test case.

[MarvinCai]

@merlimat @sijie @codelipenghui any thought on this

[linlinnn]

@MarvinCai
poll method is not a block method, it will return the head of this queue, or null if this queue is empty
pulsar-external-listener execute receivedFuture.complete(message)
and then the message is accepted by pulsar-client-internal to do final process.

Also, I see the similar lock in the method batch receive.

[linlinnn]

@eolivelli @codelipenghui @lhotari Please help review this PR, thanks.

[linlinnn]

Emm.. any feedback : ) @eolivelli @MarvinCai @merlimat @315157973

[linlinnn]

/pulsarbot run-failure-checks

[linlinnn]

/pulsarbot run-failure-checks

[linlinnn]

CICD passed. Could you please help review if you are available? @lhotari

[lhotari]

@linlinnn please rebase on latest apache/pulsar master since the problem in the master branch was fixed separately by #10250 .

[linlinnn]

@linlinnn please rebase on latest apache/pulsar master since the problem in the master branch was fixed separately by #10250 .

@lhotari Thanks. I mean could you please help review this PR.

[315157973]

IMO，it’s better to revert first, so that we can easily distinguish whether the flaky unit test is caused by my PR

[linlinnn]

IMO，it’s better to revert first, so that we can easily distinguish whether the flaky unit test is caused by my PR

+1

[codelipenghui]

Looks good to me, @linlinnn I think you can use this PR to revert #9261 first. @hangc0276 Please also help take a look.

[linlinnn]

@codelipenghui @315157973 I rebase master and revert #9261, PTAL

[eolivelli]

@hangc0276 please take a look

[MarvinCai]

LGTM, probably it is the hidden cause for many other flaky tests.

[MarvinCai]

But we should still try to improve the performance w/ a lock free solution in a separate issue ASAP.

[codelipenghui]

ping @eolivelli Please help review this PR, this issue makes many tests unstable.

[codelipenghui]

But we should still try to improve the performance w/ a lock free solution in a separate issue ASAP.

@MarvinCai Yes, we should continue to improve the performance. @315157973 also works on the lock-free solution. We'd better to revert first and then find a better solution to handle this issue.

[codelipenghui]

@linlinnn Could you please help resolve the conflict?

[eolivelli]

+1
let's merge as soon as CI passes, then we can figure out which is the best approach to improve the consumer.