Status

Current state: Under Discussion Accepted

Discussion thread: here

JIRA: Once accepted

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

...

On the level of a single partition, a relaxed offset commit check can be defined using an assignment epoch for each assigned partition and each member, which is the epoch at which the partition was assigned to that member. To fence against zombie commit requests, we can reject all offset commit requests from a member that who either does not have the partition assigned or that includes any member epoch that is smaller or equal to than the assignment epoch for that member and that partition.

...

In both cases, the zombie commit will therefore be rejected by the relaxed check.

Note that, as before, we will allow commits for partitions that are not owned by the member, if and only if the offset commit request contains the most recent broker-side member epoch. This is done to make the new rule strictly weaker than the previous check, which also allowed such commits.

Proposed Changes

Introducing an epoch per assigned partition

...

Where, for simplicity, we can assume the assignment epoch of a partition that is not assigned to that member to be Integer.maxValue.be the broker-side member epoch, to match the current behavior of accepting commits for unassigned partitions. Therefore, we allow commits for partitions that are not assigned to the member, as long as the committer uses an up-to-date member epoch, same as the current implementation.

Handling of commits without generation ID / member epoch

Requests that do not contain a valid member epoch, which can come from the admin client or a consumer with self-assigned partitions will be handled the same way as before. That is, for a non-existing groups, a simple group is created (which is not affected by this KIP). Offset commits with member epoch or generation ID -1 are unconditionally accepted of the group is empty, or the commit is transactional. 

Handling of legacy record members

...

When a static member leaves the group, the assignment epochs will be removed from set to 0 in the current assignment record. The record is then handled like a legacy record above; the assignment epoch will be initialized on load to the current member epoch, which will be -2 in this case. Partitions will When the static member rejoins, partitions will therefore be considered to be assigned "from the beginning" (i.e., from epoch 0) to the new member ID. All commits using the old memberId will be fenced, and all commits using the new memberId will be permitted (since all client-side member epochs are larger than -2), and all commits using the new memberId will be permitted.0).

Handling during protocol upgrade

If we upgrade from the classic protocol, we will set the assignment epoch of all partitions to the generation ID of the group. 

If we downgrade to the classic protocol, the assignment epochs will be abandoned, since they are not required in the classic protocol. 

Public Interfaces

ConsumerGroupCurrentMemberAssignmentValue

We add a new field, AssignmentEpochs to TopicPartitions. To keep the representation compact, we store it as an array aligned with Partitions, that is, for every item in partition there is one item in AssignmentEpochs. The field is nullable and tagged. For legacy records that do not include the assignment epochs, it will be be null, and assignment epochs for all partitions in assignedPartitions and partitionsPendingRevocation are considered to be equal to memberEpoch.

...