runLoop()

ConsumerRebalanceListener

StreamPartitionAssinger

Create StreamThread

+ Consumer(s)

+ Producer

Close StreamThread

add all records to Task RecordQueues

process records task by task
(processing order based on timestamp within each task)

-> advance "stream time"

maybePunctuate()

maybeCommit()

onPartitionsRevoke()

suspend all tasks

 - flush and commit

 - closeTopology()

subcription()

add metadata of currently assigned tasks

Brokers

assign() (Consumer Group Leader)

compute new partition/task assignment (using PartitionGrouper)

onAssign()

store assignment  

store rebalance metadata in __consumer_offsets

<groupId, (subscriptionMetadata,assignMetadata)>

onPartitionsAssing() 

- close() revoked tasks

- resume reassigned tasks -> initTopolog()

- create newly assigned tasks -> initTopology() 

  -> maybe State recovery for new tasks

while(running)

    Consumer#poll()