dajac commented on code in PR #20714:
URL: https://github.com/apache/kafka/pull/20714#discussion_r2439950813
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/OffsetMetadataManager.java:
##########
@@ -677,7 +670,20 @@ public CoordinatorResult<TxnOffsetCommitResponseData,
CoordinatorRecord> commitT
AuthorizableRequestContext context,
TxnOffsetCommitRequestData request
) throws ApiException {
- validateTransactionalOffsetCommit(context, request);
+ Group group = getGroupForTransactionalOffsetCommit(request);
Review Comment:
We don't really need the group here. Have you considered doing it the other
way around? By this, I mean having `validateTransactionalOffsetCommit` return
the `CommitPartitionValidator`? It seems we could also do the same for the
non-transactional path but we would need to re-schedule the heartbeat for
classic groups in the validation method.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/OffsetMetadataManager.java:
##########
@@ -693,6 +699,17 @@ public CoordinatorResult<TxnOffsetCommitResponseData,
CoordinatorRecord> commitT
.setPartitionIndex(partition.partitionIndex())
.setErrorCode(Errors.OFFSET_METADATA_TOO_LARGE.code()));
} else {
+ // Validate per-partition commit
+ try {
+ validator.validate(
+ topic.name(),
+ org.apache.kafka.common.Uuid.ZERO_UUID,
+ partition.partitionIndex()
+ );
+ } catch (StaleMemberEpochException ex) {
+ throw Errors.ILLEGAL_GENERATION.exception();
+ }
Review Comment:
If you follow-up on my suggestion, I wonder if we could push this into the
validator too.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
