Async Events

[sklose]

Hi,

I have a use case where I need to invalidate some state in my application when the consumer group rebalances which must complete before the rebalancing continues (in order to be compliant with the Kafka rebalance protocol and EOS semantics).

With the current API I can do

const { GROUP_JOIN } = consumer.events;
consumer.on(GROUP_JOIN, () => {
  // custom logic
});

The problem is that GROUP_JOIN is sync and my state invalidation code is inherently async.
I was thinking of submitting a PR that adds onAsync event callbacks for all the same events, but wanted to confirm with you beforehand if you have a better idea for this.

[JaapRood]

We do exactly the same thing: setup work on every rebalance before we continue processing. The trick here is to use consumer.pause and consumer.resume, which can stop and start the flow of incoming messages. As soon as the group join is received, you pause incoming messages. Once you're ready to consume again, you resume again.

Aside from that, while async event handlers are an interesting thing, they also introduce issues. The problem in particular with them here is that the Kafka client is essentially a state machine, and to function correctly as a client to the Brokers, should never really be blocked from state transitions. Async event handling does just that, so keeping them synchronous and providing other methods to control the state of the client is the simpler and preferred option.

[sklose]

@JaapRood, thanks for the quick response. Not sure if pause and resume will be sufficient for my use case. I am effectively managing my own queues of messages outside of the kafkajs runner loop and when a rebalance happens I need to wait for currently in-flight messages and clear pending items from the queue.

https://www.slideshare.net/ConfluentInc/everything-you-always-wanted-to-know-about-kafkas-rebalance-protocol-but-were-afraid-to-ask

The rebalance protocol in Kafka dictates that all consumers must stop processing messages and commit any outstanding offsets before crossing the synchronization barrier. It's mainly important for EOS, that's how double-processing of messages is avoided.

[JaapRood]

Very interesting @sklose, thanks for shedding light on that. For my client work we're dealing with exactly the same scenario as you are, but don't support EOS yet for commits stored on the Broker. We definitely have plans for that, though. Can't say that I was aware that this synchronisation barrier played an essential role in it, so will have to dive into that!

Keeping in mind I need to get into this a bit deeper, I would suggest that something of this importance would require it's own dedicated mechanic, rather than making instrumentation events asynchronous (for some of the reasons stated in my earlier response).

@tulios @Nevon this is shaping up to be pretty interesting, definitely worth a bit of your attention as well.

[sklose]

A dedicated mechanism for the synchronization barrier makes sense to me. Maybe some kind of optional (async) callback that you can configure on the client.

consumer.synchronizationBarrier(async () => {
   await myCustomBarrierCode();
});

I can work with the pause/resume functions for now. Once we settled on an approach for this I am happy to submit a PR.

FYI this is the talk I took that image from https://www.confluent.io/kafka-summit-lon19/everything-you-wanted-to-know-kafka-afraid/ (unfortunately you have to give them your email address to watch it).