Current state: Under Discussion
Discussion thread: here [Change the link from the KIP proposal email archive to your own email thread]
JIRA: here [Change the link from KAFKA-1 to your own ticket]
Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).
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.
In consumer groups based on KIP-848, when receiving an offset commit request that includes the client-side member epoch and a member ID, the group coordinator performs the check
Client-Side Member Epoch == Broker-Side Member Epoch
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.
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). In this case, the members commits will be fenced solely based on the member ID (which is not part of the group anymore). We will ignore this case in this KIP, 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 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.
This problem is particularly challenging to solve on the client side for transactional offset commits using the current Java consumer and producer implementations. The reason is that the heartbeat is sent by the consumer, but the transactional commit is initiated by the producer. The producer has no way of knowing when a consumer group heartbeat is in-flight. The member epoch is passed from the Java consumer to the Java producer using the ConsumerGroupMetadata object, which is passed into sendOffsetsToTransaction. By the time the transactional offset commit is sent, the member epoch may be stale, the broker will return an ILLEGAL_GENERATION exception. This will force the Java producer into an abortable error state, surfacing the error as a CommitFailedException to the user, the user has no other way to recover from this other than aborting the transaction.
While aborting a transaction is in principle a valid strategy for an application to recover from the ILLEGAL_GENERATION error, aborting transactions means throwing away work and restarting from an earlier point, which can hurt performance.
In this KIP, we therefore propose to relax the condition for offset commit fencing.
To derive a more relaxed check, we need to identify an epoch which separates zombie commits from commits of the current owner. As mentioned above, zombie commit requests are commit requests that include a partition, member ID and member epoch combination, so that the member owned the partition at that epoch. However, the partition has since been reassigned to a different member.
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 from zombie commit requests, we can reject all offset commit requests from a member that either does not have the partition assigned, or that includes any member epoch that is smaller or equal than the assignment epoch for that member and that partition.
Assignment Epoch <= Client-Side Member Epoch <= Broker-Side Member Epoch
Using this check, all commits of the current partition owner will be accepted, since the Client-Side Member Epoch of the current owner must always have an epoch that is larger or equal than the assignment epoch (a partition that is revoked in one epoch is never reassigned in the same epoch). All zombie commits from that member will be rejected, because if a partition was owned by the member A at Client-Side Member Epoch (which we assume for zombie commits), but it was reassigned to member B since, we have two possible cases:
Member A currently does not have the partition assigned
Member A does currently have the partition assigned, but then it must have been reassigned to member A after being assigned to member B. By KIP-848 this cannot all happen in the same epoch, so we must have Assignment Epoch > Client-Side Member Epoch.
In both cases, the zombie commit will therefore be rejected by the relaxed check.
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 AssignmentEpoch <= MemberEpoch <= TargetAssignmentEpoch <= GroupEpoch.
The AssignmentEpoch is added as a field to TopicPartitions in ConsumerGroupCurrentMemberAssignmentValue, so that it can be stored and replayed from the committed offsets topic.
We replace the current check offset commit validation check
Client-Side Member Epoch == Broker-Side Member Epoch
by
Assignment Epoch <= Client-Side Member Epoch <= Broker-Side Member Epoch
Where, for simplicity, we can assume the assignment epoch of a partition that is not assigned to that member to be Integer.maxValue.
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 null, and assignment epochs for all partitions in assignedPartitions and partitionsPendingRevocation are considered to be equal to memberEpoch.
{
"apiKey": 8,
"type": "coordinator-value",
"name": "ConsumerGroupCurrentMemberAssignmentValue",
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
...
],
"commonStructs": [
{ "name": "TopicPartitions", "versions": "0+", "fields": [
..
{ "name": "AssignmentEpochs", "versions": "0+", "nullableVersions": "0+", "taggedVersions": "0+", "tag": 0, "type": "[]int32", "default": null,
"about": "The epoch at which any partition was assigned to the member. Used to fence zombie commits requests. Of the same length as partitions. If null, all assignment epochs are considered to be equal to the member epoch." }
]}
]
} |
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.
We will extend the existing integration and unit tests to cover the new behavior.
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.
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.