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

Code before:
                    ));
                }
            });
        });

        if (!records.isEmpty()) {
            metrics.record(GroupCoordinatorMetrics.OFFSET_COMMITS_SENSOR_NAME, records.size());
        }

        return new CoordinatorResult<>(records, response);
    }

    /**
     * Handles an TxnOffsetCommit request.
     *
     * @param context The request context.
     * @param request The TxnOffsetCommit request.
     *
     * @return A Result containing the TxnOffsetCommitResponseData response and
     *         a list of records to update the state machine.
     */
    public CoordinatorResult<TxnOffsetCommitResponseData, CoordinatorRecord> commitTransactionalOffset(
        AuthorizableRequestContext context,
        TxnOffsetCommitRequestData request
    ) throws ApiException {
        Group group = getGroupForTransactionalOffsetCommit(request);

        CommitPartitionValidator validator;
        try {
            validator = group.validateOffsetCommit(
                request.memberId(),
                request.groupInstanceId(),
                request.generationId(),
                true,
                context.requestVersion()
            );
        } catch (StaleMemberEpochException ex) {
            throw Errors.ILLEGAL_GENERATION.exception();
        }

        final TxnOffsetCommitResponseData response = new TxnOffsetCommitResponseData();
        final List<CoordinatorRecord> records = new ArrayList<>();
        final long currentTimeMs = time.milliseconds();

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

            topic.partitions().forEach(partition -> {
                if (isMetadataInvalid(partition.committedMetadata())) {
                    topicResponse.partitions().add(new TxnOffsetCommitResponsePartition()

Review comment (suggestion):
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.

Code after:
                        .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
                    ));
                }
            });
        });

        if (!records.isEmpty()) {
            metrics.record(GroupCoordinatorMetrics.OFFSET_COMMITS_SENSOR_NAME, records.size());
        }

        return new CoordinatorResult<>(records, response);
    }

    /**
     * Handles an TxnOffsetCommit request.
     *
     * @param context The request context.
     * @param request The TxnOffsetCommit request.
     *
     * @return A Result containing the TxnOffsetCommitResponseData response and
     *         a list of records to update the state machine.
     */
    public CoordinatorResult<TxnOffsetCommitResponseData, CoordinatorRecord> commitTransactionalOffset(
        AuthorizableRequestContext context,
        TxnOffsetCommitRequestData request
    ) throws ApiException {
        CommitPartitionValidator validator = validateTransactionalOffsetCommit(context, request);

        final TxnOffsetCommitResponseData response = new TxnOffsetCommitResponseData();
        final List<CoordinatorRecord> records = new ArrayList<>();
        final long currentTimeMs = time.milliseconds();

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

            topic.partitions().forEach(partition -> {
                if (isMetadataInvalid(partition.committedMetadata())) {