Status

Current state: Under Discussion

...

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.

...

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 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 this point, its commits will be fenced solely based on the member ID (which is not part of the group anymore). We therefore ignore this case in this document, and only consider zombie commits from members that are still part of the group.

Downsides of the current approach

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.

Why a broker-side fix is warranted

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. 

Conceptual Design

In this KIP, we therefore propose to relax the condition for offset commit fencing.

Identifying zombies using the epoch a partition was assigned to the member

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.

...

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

Proposed Changes

Introducing Per-Member and Per-Partition Assignment Epoch

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 <= 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.

Relaxing the offset commit validation 

We replace the current check offset commit validation check

...

Where, for simplicity, we can assume the assignment epoch of a partition that is not assigned to that member to be Integer.maxValue.

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.

...

{
  "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." }
    ]}
  ]
}

Compatibility, Deprecation, and Migration Plan

• 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?

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.

References

• A similar mechanism was already implemented in KIP-1071: Streams Rebalance Protocol