[
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 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 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)
