...

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

Motivation

Fencing offset commits

In the Kafka protocol, when a consumer commits offsets or a producer tries to add offsets to a transaction, it includes its epoch/generation of the consumer group. The point of this is for the group coordinator to fence against zombie commit requests, that is, commit requests that include an offset for a partition that was since reassigned to a different member. If such a guard was not in place, a zombie offset commit may overwrite offsets of the new owner, or its offsets may be committed to the consumer offset topic but not be included in the result of the new owners offset fetch request.

...

If the check fails, it returns a STALE_MEMBER_EPOCH error for regular offset commits and a ILLEGAL_GENERATION for transactional offset commits. If the member epoch sent in the request is the current broker-side member epoch, KIP-848 guarantees that the partition cannot also be owned by a different member at the same or a larger epoch. Therefore, this is sufficient for fencing zombie commits. Note that we assume zombie commits will always contain offsets for partitions that were owned by the member at the member epoch sent in the request. Commit requests that commit offsets for partitions that are not owned by the member in that epoch, are not possible in a correct client-side implementation of the protocol.

Note that the broker-side member epoch is not the group epoch or the target assignment epoch. For details, see KIP-848. Note also that It's important to note that commits can also be fenced because a member falls out of the group (e.g. because it does not revoke partitions within the rebalance timeout). At In this pointcase, its the members commits will be fenced solely based on the member ID (which is not part of the group anymore). We therefore will ignore this case in this documentKIP, and only consider zombie commits from members that are still part of the group.

...

This fencing is, however, unnecessarily strict. Assume, for example, a member owns P1 at epoch 1. The broker-side member epoch is bumped to 2, but the member still has P1 assigned at epoch 2. The member may not learn about the new broker-side member epoch in time, and submit an offset commit commit for P1 with epoch 1. This is not a zombie commit request as define above (because P1 was not reassigned to a different member), but it will still be rejected by a KIP-848 group coordinator.

The trouble with this fencing mechanism is that it is very difficult to avoid the broker-side member epoch being bumped concurrently with an offset commit. Seen from the client-side, the broker-side member epoch may be bumped at any time while a heartbeat to the group coordinator is in-flight. To make sure the member epoch sent in an offset commit request is up-to-date would require making sure that no consumer group heartbeat request is in-flight at the same time.

Image Added

Why a broker-side fix is warranted

...

We extend the model of a consumer group with one integer per assigned partition for each member of a group. This includes both partitions directly assigned to the member, and partitions pending revocation. The assignment epoch is set to the epoch in which the partition was assigned to the member, and we have the invariant Assignment Epoch AssignmentEpoch <= MemberEpoch <= TargetAssignmentEpoch <= GroupEpoch.

...

• The records above will use a tagged field, so no record version bump needs to be done.

• If a member assignment record contains no AssignmentEpochs it will default to null. In this case, the assignment epoch of all partitions will be initialized to the current member epoch of the member. This means the original offset commit check will be used, preserving safety. The next update of the ConsumerGroupMemberCurrentAssignmentValue  record will use. 

• When rolling back, the new field will be ignored, and the original offset commit check will be used.

Test Plan

Describe in few sentences how the KIP will be tested. We are mostly interested in system tests (since unit-tests are specific to implementation details). How will we know that the implementation works as expected? How will we know nothing broke?We will extend the existing integration and unit tests to cover the new behavior.

Rejected Alternatives

• One could attempt to reach from the producer into the consumer background thread to make consumer heartbeats and transactional offset commits mutually exclusive. This would solve the problem, since member epochs are only bumped when a heartbeat is in-flight. This would be more of a workaround, and would likely require a larger change of the Java client APIs.

• We considered tracking a single RevocationEpoch  per member instead of an assignment epoch per assigned partition - which is the last epoch a partition was revoked from the member. While this would have reduced the chance of hitting the race conditions, it turned out there were still cases where you could run into spurious ILLEGAL_GENERATION errors. 

• If we use a consumer rebalance listener to always commit any open transactions in onPartitionsRevoked  and abort any open transactions in onPartitionsLost, we can actually make sure that the ILLEGAL_GENERATION can always be safely retried. The reason is that the member epoch cannot be bumped during the execution of the rebalance handler, and partitions cannot be revoked without executing the rebalance listener. So one option discussed was to simply retry the error, since we know that all partitions we are trying to commit are still assigned to us, just the member epoch was outdated. However, this would make the usage of transactions in Kafka even more complicated. Instead, we want to fix the problem on the broker side and simplify client-side usage of transactions. Furthermore, this would require significant changes in the Java producer, since it can only enter an abortable state, a fatal state or retry a request immediately. Implementing a retry after refreshing the consumer group metadata would have required a new way of interacting with the producer: We’d have to implement an “application-retriable” kind of error, that is handing back control-flow to the application without entering an error state, without retrying the commit immediately and without indicating “success” to the application.

...