Partitioned Consumer and Listener Threads

[sschepens]

I understand that Pulsar Client could and should definitely be shared among Publisher and Consumer instances to take advantage of Netty's IO threads and share them among all instances.

But there seems to be an issue particularly with PartitionedConsumerImpl because it uses a messageListener to receive messages form it's internal ConsumerImpls.

Listener are executed in a pool of configurable size, but PartitionedConsumerImpl does blocking calls, so this is definitely a problem. If, for some reason, a PartitionedConsumer stops consuming messages, it will block listener threads equal to the amount of partitions it has, or maybe even more.

I don't really have a solution, but we should definitely have a better approach to PartitionedConsumerImpl

[merlimat]

The PartitionedConsumerImpl uses a specific internal executor to handle that. See at :

https://github.com/yahoo/pulsar/blob/master/pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/PartitionedConsumerImpl.java#L47

The individual ConsumerImpl instances inside the PartitionedConsumerImpl will use that internalListenerExecutor to push messages into the common queue. If the application stops calling receive(), the common queue will get full and that will block the internal listener thread, but that will not block the IO threads, since they will just enqueue on the individual queues (which don't have size limit) and move on.

[sschepens]

Yes, I realize it will not block IO threads, but it will block other PartitionedConsumerImpl that are using the same internalListenerExecutor, a Client creates an internalListenerExecutor of size 1 by default, but even increasing the size is not a really good solution. If we intend on creating many consumers, it's not a good solution to have 1 thread per consumer, it doesn't scale.
It would be great if PartitionedConsumerImpl could be a little more smart about handling messages, but this would probably need more communication between PartitionedConsumerImpl and its internal ConsumerImpl than we currently have. For example, ConsumerImpl could try and add a message to PartitionedConsumerImpl if it doesn't succeed then it could just remain in the queue, then, PartitionedConsumerImpl could somehow notifying it's internal consumers that it has space so they pop messages from their queues.

[merlimat]

OK, that is a good point.

The key part is to leave the messages in the single topics queues, so that we can stop the broker from pushing more messages. If that can be done without recurring to blocking the intermediate listener thread, that would be great.

[merlimat]

Fixed by #106