Multiple consumers read from the same partition after rebalance

[squadgazzz]

With fs2-kafka 3.0.1 and also on 2.x.x, we faced with the issue where multiple consumers that share the same group id start reading from the same partition after rebalance. The flow looks as follows:

• Consumer A is reading from partition 1. Current offset is 1453;
• A new consumer B connects and is assigned to partition 1;
• Logs for the consumer A:

  Updating assignment with
  Assigned partitions: [mytopic.v1-2]
  Current owned partitions: [mytopic.v1-2, mytopic.v1-1]
  Added partitions (assigned - owned): [] Revoked partitions (owned - assigned): [mytopic.v1-1]
  Revoke previously assigned partitions mytopic.v1-1
  Need to revoke partitions [mytopic.v1-1] and re-join the group
  Notifying assignor about the new Assignment(partitions=[mytopic.v1-2])
  Adding newly assigned partitions:
  (Re-)joining group
  Failing OffsetCommit request since the consumer is not part of an active grou
  Successfully joined group with generation Generation{generationId=289, memberId=***protocol='cooperative-sticky'}
  Successfully synced group in generation Generation{generationId=289, memberId=***protocol='cooperative-sticky'}
  Updating assignment with
  Assigned partitions: [mytopic.v1-2]   
  Current owned partitions: [mytopic.v1-2]
  Added partitions (assigned - owned): [] Revoked partitions (owned - assigned): []
  Notifying assignor about the new Assignment(partitions=[mytopic.v1-2])
  Adding newly assigned partitions

• Consumer B starts reading from offset 1455, but consumer A continues reading from the same partition for some reason. It fails to process the event multiple times and moves to the next one, and so on.
• Sometime after, the offset for consumer A is around 1600 and for consumer B is 2000. Another rebalance happens and consumer A gets the ownership again. Since the offset was 1600 it successfully commits that offset and starts reading the event up to 2000 that were already processed by consumer B.

We have already checked the broker and found no issue. Especially the logs confirm that the broker issues the right events. There might be something with the ConsumerRebalanceListener implementation in the library.

[squadgazzz]

Might be related to the https://github.com/fd4s/fs2-kafka/issues/127

[wookievx]

We had very similar issue at my workplace that old consumer kept consuming after being revoked. From what I gathered from the linked issue and rebalance listener implementation all the streams consuming partitions that are revoked are immediately signalled to terminate. Maybe issue is due to how termination is handled in fs2.Streams/cats effect, or certain stream combinators that mess with it. I was unable to confirm that blocking the callback onRevoked actually stops other consumers from processing partition from older offset. I will probably perform experiment to check if that is the case.

This property of always being guaranteed that only one consumer is consuming a partition is very important to my use case (enabled me to store state in memory and avoid conflicts, very useful for event sourcing). In my use case I do not really care if message will be processed multiple times, as long as only one consumer is doing it at a time.

One thing ZIO kafka does differently I think is that it stores promise notifying the "actor" that partition stream finished operating, that might be a change needed here to allow the semantics described above (but it all hangs on the assumption that new assignment is blocked by revoke callback)