You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "smjn (via GitHub)" <gi...@apache.org> on 2026/08/18 08:54:10 UTC
[PR] MINOR: Idempotent initialize share group state. [kafka]
smjn opened a new pull request, #23190:
URL: https://github.com/apache/kafka/pull/23190
Currently, the ShareCoordinatorShard.initializeState does not check for dedup request resulting in overwriting the stored state, in case of an out of order initializeState. This PR remedies the situation by comparing the stateEpoch and startOffset of the incoming request with the stored state and if both these values are equal, consider the request as no-op.
This should not affect alterShareGroupOffsets as that request always brings a bumped up epoch - https://github.com/apache/kafka/pull/23147.
New tests have been added to verify the changes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Idempotent initialize share group state. [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #23190:
URL: https://github.com/apache/kafka/pull/23190#discussion_r3802635086
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorShard.java:
##########
@@ -563,15 +563,23 @@ public CoordinatorResult<InitializeShareGroupStateResponseData, CoordinatorRecor
InitializeShareGroupStateRequestData.PartitionData partitionData = topicData.partitions().get(0);
SharePartitionKey key = SharePartitionKey.getInstance(request.groupId(), topicData.topicId(), partitionData.partition());
- CoordinatorRecord record = generateInitializeStateRecord(partitionData, key);
- // build successful response if record is correctly created
+ Integer currentStateEpoch = stateEpochMap.get(key);
+ ShareGroupOffset currentState = shareStateMap.get(key);
+
InitializeShareGroupStateResponseData responseData = new InitializeShareGroupStateResponseData().setResults(
List.of(InitializeShareGroupStateResponse.toResponseInitializeStateResult(key.topicId(),
List.of(InitializeShareGroupStateResponse.toResponsePartitionResult(
key.partition()))
))
);
+ if (currentStateEpoch != null && currentStateEpoch == partitionData.stateEpoch() &&
+ currentState != null && currentState.startOffset() == partitionData.startOffset()) {
Review Comment:
Should they not have >= check to see if the request is out of order?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Idempotent initialize share group state. [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 merged PR #23190:
URL: https://github.com/apache/kafka/pull/23190
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Idempotent initialize share group state. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #23190:
URL: https://github.com/apache/kafka/pull/23190#discussion_r3827492086
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorShard.java:
##########
@@ -563,15 +563,23 @@ public CoordinatorResult<InitializeShareGroupStateResponseData, CoordinatorRecor
InitializeShareGroupStateRequestData.PartitionData partitionData = topicData.partitions().get(0);
Review Comment:
Yes, sounds reasonable - https://issues.apache.org/jira/browse/KAFKA-20972
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Idempotent initialize share group state. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #23190:
URL: https://github.com/apache/kafka/pull/23190#discussion_r3803411626
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorShard.java:
##########
@@ -563,15 +563,23 @@ public CoordinatorResult<InitializeShareGroupStateResponseData, CoordinatorRecor
InitializeShareGroupStateRequestData.PartitionData partitionData = topicData.partitions().get(0);
SharePartitionKey key = SharePartitionKey.getInstance(request.groupId(), topicData.topicId(), partitionData.partition());
- CoordinatorRecord record = generateInitializeStateRecord(partitionData, key);
- // build successful response if record is correctly created
+ Integer currentStateEpoch = stateEpochMap.get(key);
+ ShareGroupOffset currentState = shareStateMap.get(key);
+
InitializeShareGroupStateResponseData responseData = new InitializeShareGroupStateResponseData().setResults(
List.of(InitializeShareGroupStateResponse.toResponseInitializeStateResult(key.topicId(),
List.of(InitializeShareGroupStateResponse.toResponsePartitionResult(
key.partition()))
))
);
+ if (currentStateEpoch != null && currentStateEpoch == partitionData.stateEpoch() &&
+ currentState != null && currentState.startOffset() == partitionData.startOffset()) {
Review Comment:
@apoorvmittal10
Thanks for the review. The validation for < is done in `maybeGetInitializeStateError` before this executes.
The case we are concerned here is mainly for initializeState retries which contain exactly same information. Assume 2 initializeState calls i1, i2. Due to network issue say i1 is executed on share coord correctly but the response does not arrive at the persister. Persister sends i2 (retry) - this is exactly the type of case we want ot check here. Same thing if i2 completes but i1 arrives out of order (stateEpochs are same). This covers ==
For > case, the request could be from alter offsets for which we want to overwrite state.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Idempotent initialize share group state. [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #23190:
URL: https://github.com/apache/kafka/pull/23190#discussion_r3804331618
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorShard.java:
##########
@@ -563,15 +563,23 @@ public CoordinatorResult<InitializeShareGroupStateResponseData, CoordinatorRecor
InitializeShareGroupStateRequestData.PartitionData partitionData = topicData.partitions().get(0);
Review Comment:
Out of curiosity. Do we plan to make `initializeState` support multiple keys in the future? If not, should `initializeState` accept `groupId`, `topicId`, and `PartitionData` directly instead of `InitializeShareGroupStateRequestData`?
It brings the following benefits:
1. we don't need to construct construct single-element `InitializeShareGroupStateRequestData` objects again in the service layer
2. we can avoid the hardcoded .get(0) calls in the shard layer
WDYT?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Idempotent initialize share group state. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #23190:
URL: https://github.com/apache/kafka/pull/23190#discussion_r3803411626
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorShard.java:
##########
@@ -563,15 +563,23 @@ public CoordinatorResult<InitializeShareGroupStateResponseData, CoordinatorRecor
InitializeShareGroupStateRequestData.PartitionData partitionData = topicData.partitions().get(0);
SharePartitionKey key = SharePartitionKey.getInstance(request.groupId(), topicData.topicId(), partitionData.partition());
- CoordinatorRecord record = generateInitializeStateRecord(partitionData, key);
- // build successful response if record is correctly created
+ Integer currentStateEpoch = stateEpochMap.get(key);
+ ShareGroupOffset currentState = shareStateMap.get(key);
+
InitializeShareGroupStateResponseData responseData = new InitializeShareGroupStateResponseData().setResults(
List.of(InitializeShareGroupStateResponse.toResponseInitializeStateResult(key.topicId(),
List.of(InitializeShareGroupStateResponse.toResponsePartitionResult(
key.partition()))
))
);
+ if (currentStateEpoch != null && currentStateEpoch == partitionData.stateEpoch() &&
+ currentState != null && currentState.startOffset() == partitionData.startOffset()) {
Review Comment:
@apoorvmittal10
Thanks for the review. The validation is done in `maybeGetInitializeStateError` before this executes. That covers the case you mentioned.
The case we are concerned here is mainly for initializeState retries which contain exactly same information. Assume 2 initializeState calls i1, i2. Due to network issue say i1 is executed on share coord correctly but the response does not arrive at the persister. Persister sends i2 (retry) - this is exactly the type of case we want ot check here. Same thing if i2 completes but i1 arrives out of order (stateEpochs are same).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Idempotent initialize share group state. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #23190:
URL: https://github.com/apache/kafka/pull/23190#discussion_r3827492086
##########
share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorShard.java:
##########
@@ -563,15 +563,23 @@ public CoordinatorResult<InitializeShareGroupStateResponseData, CoordinatorRecor
InitializeShareGroupStateRequestData.PartitionData partitionData = topicData.partitions().get(0);
Review Comment:
Yes, sounds reasonable - https://issues.apache.org/jira/browse/KAFKA-20972
https://github.com/apache/kafka/pull/23224
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org