[
https://issues.apache.org/jira/browse/KAFKA-20910?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Sushant Mahajan updated KAFKA-20910:
------------------------------------
Description:
The `GroupCoordinatorService.alterShareGroupOffsets` impl which eventually
executes `GroupMetadataManager.alterShareGroupOffsets` directly runs the share
partition initialize logic without bumping the group epoch as highlighted in
the original KIP spec.
Excerpt from the
[kip-932|https://cwiki.apache.org/confluence/spaces/KAFKA/pages/255070434/KIP-932+Queues+for+Kafka]
{{Only empty share groups support this operation. The group coordinator bumps
the group epoch, writes a {*}ShareGroupMetadata{*}, and sends an
{{InitializeShareGroupState}} request to the share coordinator. The share
coordinator writes a *ShareSnapshot* record with the new state epoch to the
{{__share_group_state}} topic. If the partition was not previously
initialized, the group coordinator writes an updated
*ShareGroupStatePartitionMetadata* record.}}
This is not just about adhering to the spec but about the share coordinator's
ability to guarantee idempotency and differentiating between different kind of
requests. For example, without this fix, the share coordinator cannot make out
what to do for out order initializeState requests. This is because
alterShareGroupOffsets is an admin command which must be honored. If they both
use the same stateEpoch, some older network delayed initializeState might
overwrite the what alter did and if we just do a no-op return when stateEpoch
is the same - we might not honor the alter request.
was:
The `GroupCoordinatorService.alterShareGroupOffsets` impl which eventually
executes `GroupMetadataManager.alterShareGroupOffsets` directly runs the share
partition initialize logic without bumping the group epoch as highlighted in
the original KIP spec.
Excerpt from the
[kip-932|https://cwiki.apache.org/confluence/spaces/KAFKA/pages/255070434/KIP-932+Queues+for+Kafka]
{{Only empty share groups support this operation. The group coordinator bumps
the group epoch, writes a {*}ShareGroupMetadata{*}, and sends an
{{InitializeShareGroupState}} request to the share coordinator. The share
coordinator writes a *ShareSnapshot* record with the new state epoch to the
{{__share_group_state}} topic. If the partition was not previously
initialized, the group coordinator writes an updated
*ShareGroupStatePartitionMetadata* record.}}
This is not just about adhering to the spec but about the the share
coordinator's ability to guarantee idempotency and differentiating between
different kind of requests. For example, without this fix, the share
coordinator cannot make out what to do for out order initializeState requests.
This is because alterShareGroupOffsets is an admin command which must be
honored. If they both use the same stateEpoch, some older network delayed
initializeState might overwrite the what alter did and if we just do a no-op
return when stateEpoch is the same - we might not honor the alter request.
> alterShareGroupOffsets impl in GroupCoordinatorService does not follow the
> KIP-932 spec.
> ----------------------------------------------------------------------------------------
>
> Key: KAFKA-20910
> URL: https://issues.apache.org/jira/browse/KAFKA-20910
> Project: Kafka
> Issue Type: Bug
> Reporter: Sushant Mahajan
> Assignee: Sushant Mahajan
> Priority: Major
>
> The `GroupCoordinatorService.alterShareGroupOffsets` impl which eventually
> executes `GroupMetadataManager.alterShareGroupOffsets` directly runs the
> share partition initialize logic without bumping the group epoch as
> highlighted in the original KIP spec.
> Excerpt from the
> [kip-932|https://cwiki.apache.org/confluence/spaces/KAFKA/pages/255070434/KIP-932+Queues+for+Kafka]
>
> {{Only empty share groups support this operation. The group coordinator bumps
> the group epoch, writes a {*}ShareGroupMetadata{*}, and sends an
> {{InitializeShareGroupState}} request to the share coordinator. The share
> coordinator writes a *ShareSnapshot* record with the new state epoch to the
> {{__share_group_state}} topic. If the partition was not previously
> initialized, the group coordinator writes an updated
> *ShareGroupStatePartitionMetadata* record.}}
>
> This is not just about adhering to the spec but about the share coordinator's
> ability to guarantee idempotency and differentiating between different kind
> of requests. For example, without this fix, the share coordinator cannot make
> out what to do for out order initializeState requests. This is because
> alterShareGroupOffsets is an admin command which must be honored. If they
> both use the same stateEpoch, some older network delayed initializeState
> might overwrite the what alter did and if we just do a no-op return when
> stateEpoch is the same - we might not honor the alter request.
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
