@Override
    public DescribeProducersRequest.Builder buildBatchedRequest(
        int brokerId,
        Set<TopicPartition> topicPartitions
    ) {
        DescribeProducersRequestData request = new DescribeProducersRequestData();
        DescribeProducersRequest.Builder builder = new DescribeProducersRequest.Builder(request);

        CollectionUtils.groupPartitionsByTopic(
            topicPartitions,
            builder::addTopic,
            (topicRequest, partitionId) -> topicRequest.partitionIndexes().add(partitionId)
        );

        return builder;
    }
@Test
    public void testBuildRequest() {
        Set<TopicPartition> topicPartitions = mkSet(
            new TopicPartition("foo", 5),
            new TopicPartition("bar", 3),
            new TopicPartition("foo", 4)
        );

        DescribeProducersHandler handler = newHandler(
            new DescribeProducersOptions()
        );

        int brokerId = 3;
        DescribeProducersRequest.Builder request = handler.buildBatchedRequest(brokerId, topicPartitions);

        List<DescribeProducersRequestData.TopicRequest> topics = request.data.topics();

        assertEquals(mkSet("foo", "bar"), topics.stream()
            .map(DescribeProducersRequestData.TopicRequest::name)
            .collect(Collectors.toSet()));

        topics.forEach(topic -> {
            Set<Integer> expectedTopicPartitions = "foo".equals(topic.name()) ?
                mkSet(4, 5) : mkSet(3);
            assertEquals(expectedTopicPartitions, new HashSet<>(topic.partitionIndexes()));
        });
    }