Repository: apache/kafka
File: group-coordinator/src/main/java/org/apache/kafka/coordinator/group/OffsetMetadataManager.java (Java)

Code before:
        final OffsetCommitResponseData response = new OffsetCommitResponseData();
        final List<CoordinatorRecord> records = new ArrayList<>();
        final long currentTimeMs = time.milliseconds();
        final OptionalLong expireTimestampMs = expireTimestampMs(request.retentionTimeMs(), currentTimeMs);

        request.topics().forEach(topic -> {
            final OffsetCommitResponseTopic topicResponse = new OffsetCommitResponseTopic()
                .setTopicId(topic.topicId())
                .setName(topic.name());
            response.topics().add(topicResponse);

            topic.partitions().forEach(partition -> {
                if (isMetadataInvalid(partition.committedMetadata())) {
                    topicResponse.partitions().add(new OffsetCommitResponsePartition()
                        .setPartitionIndex(partition.partitionIndex())
                        .setErrorCode(Errors.OFFSET_METADATA_TOO_LARGE.code()));
                } else {
                    // Validate per-partition commit if needed
                    if (requiresPartitionValidation) {
                        group.validateOffsetCommitForPartition(
                            request.memberId(),
                            request.generationIdOrMemberEpoch(),
                            topic.name(),
                            topic.topicId(),
                            partition.partitionIndex()
                        );
                    }

                    log.debug("[GroupId {}] Committing offsets {} for partition {}-{}-{} from member {} with leader epoch {}.",
                        request.groupId(), partition.committedOffset(), topic.topicId(), topic.name(), partition.partitionIndex(),
                        request.memberId(), partition.committedLeaderEpoch());

                    topicResponse.partitions().add(new OffsetCommitResponsePartition()
                        .setPartitionIndex(partition.partitionIndex())
                        .setErrorCode(Errors.NONE.code()));

                    final OffsetAndMetadata offsetAndMetadata = OffsetAndMetadata.fromRequest(
                        topic.topicId(),
                        partition,
                        currentTimeMs,
                        expireTimestampMs
                    );

                    records.add(GroupCoordinatorRecordHelpers.newOffsetCommitRecord(
                        request.groupId(),
                        topic.name(),
                        partition.partitionIndex(),
                        offsetAndMetadata
                    ));
                }
            });

Review comment (performance):
This may be a bit inefficient because we have to lookup the member for every partition. I wonder whether we could do something as follow.

Let's define an OffsetPartitionValidator. We could find a better name.

```
OffsetPartitionValidator {
  void validate(name, topicId, partitionId)
}
```

At the group level, we can have a method which return the a concrete OffsetPartitionValidator depending on the member epoch / group type. If the epoch matches the expected epoch, the validator is basically a no op. Otherwise, we return one which does the partition level check while caching the reference to the member.

```
Group {
  OffsetPartitionValidator offsetValidator(memberId, memberEpoch)
}
```

This would make the validation logic in this class kind of agnostic of the two level validation.

Have you considered something like this? What do you think about it? 

Code after:
     * Handles an OffsetCommit request.
     *
     * @param context The request context.
     * @param request The OffsetCommit request.
     *
     * @return A Result containing the OffsetCommitResponseData response and
     *         a list of records to update the state machine.
     */
    public CoordinatorResult<OffsetCommitResponseData, CoordinatorRecord> commitOffset(
        AuthorizableRequestContext context,
        OffsetCommitRequestData request
    ) throws ApiException {
        CommitPartitionValidator validator = validateOffsetCommit(context, request);

        final OffsetCommitResponseData response = new OffsetCommitResponseData();
        final List<CoordinatorRecord> records = new ArrayList<>();
        final long currentTimeMs = time.milliseconds();
        final OptionalLong expireTimestampMs = expireTimestampMs(request.retentionTimeMs(), currentTimeMs);

        request.topics().forEach(topic -> {
            final OffsetCommitResponseTopic topicResponse = new OffsetCommitResponseTopic()
                .setTopicId(topic.topicId())
                .setName(topic.name());
            response.topics().add(topicResponse);

            topic.partitions().forEach(partition -> {
                if (isMetadataInvalid(partition.committedMetadata())) {
                    topicResponse.partitions().add(new OffsetCommitResponsePartition()
                        .setPartitionIndex(partition.partitionIndex())
                        .setErrorCode(Errors.OFFSET_METADATA_TOO_LARGE.code()));
                } else {
                    // Validate commit per-partition
                    validator.validate(
                        topic.name(),
                        topic.topicId(),
                        partition.partitionIndex()
                    );

                    log.debug("[GroupId {}] Committing offsets {} for partition {}-{}-{} from member {} with leader epoch {}.",
                        request.groupId(), partition.committedOffset(), topic.topicId(), topic.name(), partition.partitionIndex(),
                        request.memberId(), partition.committedLeaderEpoch());

                    topicResponse.partitions().add(new OffsetCommitResponsePartition()
                        .setPartitionIndex(partition.partitionIndex())
                        .setErrorCode(Errors.NONE.code()));

                    final OffsetAndMetadata offsetAndMetadata = OffsetAndMetadata.fromRequest(
                        topic.topicId(),
                        partition,
                        currentTimeMs,
                        expireTimestampMs