Page History

...

1. For every consumer: before sending the join-group request, change the behavior as follows based on the join-group triggering event:
   1. If subscription has changed, or topic metadata has changed: do NOT revoke any partitions or call rebalance-listener; instead just encode the current assigned partitions as part of the Subscription.
   2. If received REBALANCE_IN_PROGRESS from heartbeat response or commit response: same as a) above.
   3. If received MEMBER_ID_REQUIRED from join-group request: same as a) above.
   4. If received UNKNOWN_MEMBER_ID or ILLEGAL_GENERATION from join-group / sync-group / commit / heartbeat response: reset generation / clear member-id correspondingly, emigrate the partitions with rebalance listener and then re-join group with empty assigned partition.
   5. If received assignment from previous rebalance's sync-group response contains error code NEEDS_REJOIN, revoke partitions as required before sending the join-group request with newly formed assigned partitions.
2. For the leader: after getting the received subscription topics, as well as the assigned-partitions, do the following:
   
   1. Take the cluster metadata and infer all the topic Collect the partitions that are targeted for assignmentclaimed as currently owned from the subscriptions; let's call it totalowned-known-partitions.
   2. Call the registered assignor of the selected protocol, passing in total-known-partitions as part of the cluster metadata and get the returned assignment; let's call it newly-the returned assignment assigned-partitions. Note the set of it should be equal to total-knowncould be different from owned-partitions.
   3. Compare the newly-assignedowned-partitions with assigned-partitions and generate three exclusive sub-sets:
      1. Intersection(totalowned-partitions, assigned-partitions). These are partitions that are still owned by some members, and some of them may be now allocated for new members. Let's call it prevmaybe-assignedrevoking-partitions.
      2. Minus(totalassigned-partitions, assignedowned-partitions). These are partitions that are not previously owned by any one. This set is non-empty when its previous owner is on older version and hence revoked them already before joining, or a partition is revoked in previous rebalance by the new versioned member and hence not in any assigned partitions, or it is a newly created partition due to add-partitions. Let's call it notready-to-assignedmigrate-partitions.
      3. Minus(assignedowned-partitions, totalassigned-partitions). These are partitions that are not known from the cluster metadata does not exist in assigned partitions, but are claimed to be owned by the members. It is non-empty if some topics are deleted, or if the leader's metadata is not refreshedstale (and hence the generated assignment does not have those topics), or if the previous leader has created some topics in its assignor that are not in the cluster yet (consider the Streams case). Let's call it unknown-but-owned-partitions.
         
         
   4. For notready-to-assignedmigrate-partitions, it is safe to move them to the new member immediately since we know no one owns it before, and hence we can encode the owner from the newly-assigned-partitions directly.
   5. For unknown-but-owned-partitions, it is also safe to just do nothing but just give them back to whoever claimed to be their owners by encoding them directly as well. If this is due to topic metadata update them , then a later rebalance will be triggered anyways.
   6. For prevmayby-assignedrevoking-partitions, check if the owner has changed, if . If yes, encode it to the old owner in revoked-partitions but NOT encode to the assigned-partitions to the new owner, set the error-code to NEED_REJOIN for all the members (this is for fast re-join and rebalance).
3. For every consumer: after received the sync-group request, do the following:
   1. Check that the newly assigned-partitions is a superset of Minus(assigned-partitions, revoked-partitions). This is because under cooperative rebalance, no partitions should be migrated directly before being revoked first.
      If this is not true, it means either bugs or incompatible members are forming the group (for incompatible members, see below), and hence the consumer should log a fatal error and shutdown gracefully.
   2. Update the newly assigned-partitions, and for those newly added  partitions, call the rebalance-listener — this is the same as the current logic.
   3. If revoked partitions is not empty, remove those partitions by calling the rebalance-listener.
   4. Check the error code:
      1. If it is NONE, complete.
      2. If it is NEEDS_REJOIN, immediately send another join group request with the updated assigned partitions following step 1.e) above. 

...

Below is a more illustrative of the different set of partitions and their assignment logic:

Image Modified

No changes required from the broker side, since this logic change is completely wrapped inside the consumer protocol / coordinator implementation itself, and to brokers it is just the same as previous version's rebalances.

...