The problem
When rebalance-safe-commits is enabled, during a rebalance, zio-kafka's rebalance listener waits for revoked streams to end, and the last offset they consumed to be committed. Partitions that are not revoked in the rebalance, will continue to process records. After the rebalance, the consumer might still be processing records that were received from before the rebalance. In cases where the epoch of the consumed records is important (e.g. transactional kafka), this causes problems.
The only (known) partition assignor to not revoke partitions (in case it will be assigned to the same consumer) is the CooperativeStickyAssignor. Therefore, in #1597, we discourage use of the CooperativeStickyAssignor as a workaround for this problem.
Solution
We want to change the behavior of the zio-kafka rebalance listener such that it:
- for revoked partitions: waits for a commit of the last record that was consumed by the partition stream, and
- for non-revoked partitions, waits for a commit of the last record that zio-kafka received in the currently ending epoch.
The rebalance listener no longer has to wait for any streams to end. To make this possible, we need to start tracking, for each partition, what is the last offset of the current epoch. For ending streams, we no longer track the latest offset given to the stream, but only the last offset given to the stream (or if it never received any records, a marker that it ended without consuming).
Consequences
With this in place, rebalance-safe-commits will be compatible with the CooperativeStickyAssignor and it is no longer necessary for zio-kafka to override the default partition assignor.
During a rebalance, zio-kafka does not pass pre-fetched records to ending partition streams (they are dropped). However, it must continue passing pre-fetched records to non-ending partition streams. This increases the rebalance duration. Therefore, (after this change) the CooperativeStickyAssignor is only suitable for high throughput applications. Applications that have high processing time per record still need to use another partition assignor.
Related:
The problem
When rebalance-safe-commits is enabled, during a rebalance, zio-kafka's rebalance listener waits for revoked streams to end, and the last offset they consumed to be committed. Partitions that are not revoked in the rebalance, will continue to process records. After the rebalance, the consumer might still be processing records that were received from before the rebalance. In cases where the epoch of the consumed records is important (e.g. transactional kafka), this causes problems.
The only (known) partition assignor to not revoke partitions (in case it will be assigned to the same consumer) is the CooperativeStickyAssignor. Therefore, in #1597, we discourage use of the CooperativeStickyAssignor as a workaround for this problem.
Solution
We want to change the behavior of the zio-kafka rebalance listener such that it:
The rebalance listener no longer has to wait for any streams to end. To make this possible, we need to start tracking, for each partition, what is the last offset of the current epoch. For ending streams, we no longer track the latest offset given to the stream, but only the last offset given to the stream (or if it never received any records, a marker that it ended without consuming).
Consequences
With this in place, rebalance-safe-commits will be compatible with the CooperativeStickyAssignor and it is no longer necessary for zio-kafka to override the default partition assignor.
During a rebalance, zio-kafka does not pass pre-fetched records to ending partition streams (they are dropped). However, it must continue passing pre-fetched records to non-ending partition streams. This increases the rebalance duration. Therefore, (after this change) the CooperativeStickyAssignor is only suitable for high throughput applications. Applications that have high processing time per record still need to use another partition assignor.
Related: