How to coordinate over multiple kafka subscribers/consumers in a consumer group when partitions and workers > 1

[hmvp]

So I have an application where I want to be able to reread the kafka topic to recreate the resulting database. This would be a manually triggered action...

I have all the code to make that work. Part of it needs falling back to aiokafka to do the actual reset, but that is not my issue, it works fine when workers == 1.
In order to make this work all consumers in the group need to have their offset set back to the beginning.
Due to how kafka partitions work I cannot do it from outside the consumers.

This I can do easily within a single consumer but I need a way to trigger it in all consumers, thus triggering on a kafka message in another topic does not work since I cannot make sure every consumer gets that message.

Furthermore I need to check if all consumers have finished processing the whole stream and do something else. This is also doable by measuring the lag (highwater - position = lag) and do something when that is lower than threshold. But I have not found a usable way to make sure that is not done X times (for each worker).

I looked at anyio synchronization primitives for this, but have not yet found a usable way to do this..

Anyone ideas?

[Lancetnik]

I am not sure, how consumer groups should works in a concurrent mode (with max_workers), this reason I limited concurrent subscription in autocommit (ack-first) mode only. That you created manual commit & max_workers option looks like a bug and misconfiguration for me. Anyway, my main broker specialization is RMQ and I am not experted enough to help in complex Kafka cases, but I'll try to find a person to consult with.

[hmvp]

Ah yes the max_worker settings does not work. However I used the CLI which basically spins up a number of processes. Or in a similar way, if you use kubernetes for example and you use multiple replicas.

This last part basically suggest the correct solution here: You need bigger guns to do the synchronization.. And since a faststream consumer/broker is already connected to a message bus, there is a solution available...

I landed on using a different topic/subscriber on the same Kafka instance/broker where the consumers are all in their own consumer group. This means they all get the same message.