@Override
    public MetadataRequest.Builder buildRequest(Set<TopicPartition> partitions) {
        MetadataRequestData request = new MetadataRequestData();
        request.setAllowAutoTopicCreation(false);
        partitions.stream().map(TopicPartition::topic).distinct().forEach(topic ->
            request.topics().add(new MetadataRequestData.MetadataRequestTopic().setName(topic))
        );
        return new MetadataRequest.Builder(request);
    }
@Test
    public void testBuildLookupRequest() {
        Set<TopicPartition> topicPartitions = mkSet(
            new TopicPartition("foo", 0),
            new TopicPartition("bar", 0),
            new TopicPartition("foo", 1),
            new TopicPartition("baz", 0)
        );

        PartitionLeaderStrategy strategy = newStrategy();

        MetadataRequest allRequest = strategy.buildRequest(topicPartitions).build();
        assertEquals(mkSet("foo", "bar", "baz"), new HashSet<>(allRequest.topics()));
        assertFalse(allRequest.allowAutoTopicCreation());

        MetadataRequest partialRequest = strategy.buildRequest(
            topicPartitions.stream().filter(tp -> tp.topic().equals("foo")).collect(Collectors.toSet())
        ).build();
        assertEquals(mkSet("foo"), new HashSet<>(partialRequest.topics()));
        assertFalse(partialRequest.allowAutoTopicCreation());
    }