Fix possible deadlock when AckWithResponse is true due to queueCh is full

[BewareMyPower]

Motivation

When AckWithResponse is enabled, a deadlock could happen easily if queueCh and messagesCh of partitionConsumer are full. In this case, if the consumer receives new messages from broker, MessageReceived will be blocked at

pulsar-client-go/pulsar/consumer_partition.go

Line 1376 in 9366a0e

pc.queueCh <- msg

The stacks could be:

3  0x0000000100ea83c0 in github.com/apache/pulsar-client-go/pulsar.(*partitionConsumer).MessageReceived
   at ./consumer_partition.go:1376
4  0x0000000100b1c73c in github.com/apache/pulsar-client-go/pulsar/internal.(*connection).handleMessage
   at ./internal/connection.go:751
5  0x0000000100b1ab7c in github.com/apache/pulsar-client-go/pulsar/internal.(*connection).internalReceivedCommand
   at ./internal/connection.go:590

As shown in the stacks above, the connection.run() goroutine is blocked so that it could not handle any new command anymore, including the ACK response. Then the ACK related method will fail with "request timed out". The deadlock cannot be resolved unless the consumer peeks new messages again to make messagesCh not full, then queueCh will move messages to messagesCh so that queueCh will be not full.

The root cause is that queueCh is a buffered channel that has a fixed size of ReceiverQueueSize. However, the broker could dispatch more messages than the ReceiverQueueSize because the permits in Flow requests only limits the number of entries to read, not the number of messages. Hence this issue could be easily reproduced by reducing the ReceiverQueueSize and sending many messages with a great batch size. See how TestAckResponseNotBlocked reproduces this issue.

Modifications

Add a list of messages (pendingMessages) to support queueing unlimited number of messages for partitionedConsumer. The flow control is actually controlled by the availablePermits and related Flow requests so that the queue size won't be too large (it depends on the batch size).

Add two channels (queueInCh and queueOutCh) for the following loops:

• Run a background goroutine to send messages from queueInCh -> pendingMessages -> queueOutCh
• In MessageReceived, send the received message to queueInCh so that the message will be queued to pendingMessages. It could never be blocked.
• In dispatcher(), poll messages from queueOutCh, which polls the 1st message from pendingMessages.

The background goroutine will exist by close(pc.closeCh) when the consumer is closed. It happens after completing the CloseConsumer RPC in internalClose so there should not be a MessageReceived call that sends messages to the closed queueInCh.

Add TestAckResponseNotBlocked to verify it works with a very small receiver queue size config.

[fracasula]

I reviewed this PR because it was not in a draft state when I started.

I wanted to comment also that perhaps using two channels queueInCh and queueOutCh might help make the code more understandable and easier to maintain.

What it's being proposed here is to use a more complex object with an embedded list, 3 channels and its own go routine running in the background.

[BewareMyPower]

Hi @fracasula, thanks for your comments and I've addressed all of them. You can check out the design from updated Modifications part in the PR description.

@shibd @RobertIndie Please also take a look.

[shibd]

hi, @BewareMyPower I put our discuss on here.

Before #1283, There won't be the issue mentioned in this PR because we can ensure that the pulled messages will never overflow queueCh. Previously, it was an array, and if an entry had batch messages, they were stored as an array.

Furthermore, the availablePermits mechanism will ensure that the number of pulled entries does not exceed the size set by pendingQueueSize.

[shibd]

I have revert #1283 and copy unit test from this PR: #1311

This is an alternative solution to the issue.

The current PR offers a method similar to an unbounded channel.

The PR #1311 maintains the initial behavior, using availablePermits to ensure the channel is never filled.

We can compare which approach is more user-friendly. From my perspective, #1311 is simpler and the code is more readable.