Use semaphore-like permits for some actor operations

[bplommer]

This was supposed to be a demo but I got a bit carried away - inspired by #903

[bplommer]

Looking into this, some actor operations delegate to other actor operations, pushing them to the back of the queue - I wonder if that could be the cause of some of the odd race conditions we've seen?

No I don't think so - the only code in the actor that pushes requests to the actor is for offset commits, which aren't themselves invoked via the actor.

[bplommer]

Oops, I broke something somewhere along the line 😬

[LMnet]

I checked these changes briefly. They may be good by itself, but I talked about a different approach: I wanted to completely get rid of a KafkaConsumerActor. We don't need it actually. We could call а java consumer from the fs2.kafka.KafkaConsumer through semaphore without KafkaConsumerActor.

To create polling we could run a fiber that uses the same semaphore and the same java consumer and run this fiber in the background.

[bplommer]

I agree that would be preferable as an end state, but I think the changes in this PR make direct progress towards it: the code that calls permit could directly switch to using a cats-effect semaphore.

One thing about using a single semaphore that may be an issue is that it would be strictly FIFO rather than the current approach that distinguishes polls from other requests - that said, I’m not sure the current approach is actually correct. What I thought we were currently doing, and what seems right, is to always handle all pending requests before doing a poll, but what actually seems to happen is that once the request queue is drained the actor will block until the next poll, and only after the next poll will it process requests that happened after the queue was drained. @vlovgr can you clarify what the intended behaviour is?

[vlovgr]

Ideally, we should process all fetch requests before polling, since partitions without fetch requests will remain paused when polling (as you note, we aren't really doing exactly that today). If it makes a difference in practice versus just processing everything FIFO is difficult to say. The only contract we really have to adhere to is to ensure poll is called at least every max.poll.interval.ms.

[bplommer]

Is there any objection to merging these changes as a step towards using a semaphore for everything?