@Override
    public OffsetCommitRequest.Builder buildBatchedRequest(
        int coordinatorId,
        Set<CoordinatorKey> groupIds
    ) {
        validateKeys(groupIds);

        Map<String, OffsetCommitRequestTopic> offsetData = new HashMap<>();
        offsets.forEach((topicPartition, offsetAndMetadata) -> {
            OffsetCommitRequestTopic topic = offsetData.computeIfAbsent(
                topicPartition.topic(),
                key -> new OffsetCommitRequestTopic().setName(topicPartition.topic())
            );

            topic.partitions().add(new OffsetCommitRequestPartition()
                .setCommittedOffset(offsetAndMetadata.offset())
                .setCommittedLeaderEpoch(offsetAndMetadata.leaderEpoch().orElse(-1))
                .setCommittedMetadata(offsetAndMetadata.metadata())
                .setPartitionIndex(topicPartition.partition()));
        });

        OffsetCommitRequestData data = new OffsetCommitRequestData()
            .setGroupId(groupId.idValue)
            .setTopics(new ArrayList<>(offsetData.values()));

        return new OffsetCommitRequest.Builder(data);
    }
@Test
    public void testBuildRequest() {
        AlterConsumerGroupOffsetsHandler handler = new AlterConsumerGroupOffsetsHandler(groupId, partitions, logContext);
        OffsetCommitRequest request = handler.buildBatchedRequest(-1, singleton(CoordinatorKey.byGroupId(groupId))).build();
        assertEquals(groupId, request.data().groupId());
        assertEquals(2, request.data().topics().size());
        assertEquals(2, request.data().topics().get(0).partitions().size());
        assertEquals(offset, request.data().topics().get(0).partitions().get(0).committedOffset());
    }