[
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 about out of order initializeState requests. This is because
alterShareGroupOffsets is an admin command which must be honored. If both
initialize and alter (internally initialize) use the same stateEpoch, some
older network delayed initializeState might overwrite 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 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 about out of order initializeState requests. This is because
> alterShareGroupOffsets is an admin command which must be honored. If both
> initialize and alter (internally initialize) use the same stateEpoch, some
> older network delayed initializeState might overwrite 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)
