Fetcher { public Map<String, List<PartitionInfo>> getAllTopicMetadata(long timeout) { return getTopicMetadata(MetadataRequest.allTopics(), timeout); } Fetcher(ConsumerNetworkClient client,
        int minBytes,
        int maxWaitMs,
        int fetchSize,
        int maxPollRecords,
        boolean checkCrcs,
        Deserializer<K> keyDeserializer,
        Deserializer<V> valueDeserializer,
        Metadata metadata,
        SubscriptionState subscriptions,
        Metrics metrics,
        String metricGrpPrefix,
        Time time,
        long retryBackoffMs); void sendFetches(); Map<TopicPartition, List<ConsumerRecord<K, V>>> fetchedRecords(); void updateFetchPositions(Set<TopicPartition> partitions); Map<String, List<PartitionInfo>> getAllTopicMetadata(long timeout); Map<String, List<PartitionInfo>> getTopicMetadata(MetadataRequest request, long timeout);  }
@Test public void testGetAllTopics() { client.prepareResponse(newMetadataResponse(topicName, Errors.NONE).toStruct()); Map<String, List<PartitionInfo>> allTopics = fetcher.getAllTopicMetadata(5000L); assertEquals(cluster.topics().size(), allTopics.size()); }
@Test public void testGetAllTopicsDisconnect() { client.prepareResponse(null, true); client.prepareResponse(newMetadataResponse(topicName, Errors.NONE).toStruct()); Map<String, List<PartitionInfo>> allTopics = fetcher.getAllTopicMetadata(5000L); assertEquals(cluster.topics().size(), allTopics.size()); }
@Test(expected = TimeoutException.class) public void testGetAllTopicsTimeout() { fetcher.getAllTopicMetadata(50L); }
@Test public void testGetAllTopicsUnauthorized() { client.prepareResponse(newMetadataResponse(topicName, Errors.TOPIC_AUTHORIZATION_FAILED).toStruct()); try { fetcher.getAllTopicMetadata(10L); fail(); } catch (TopicAuthorizationException e) { assertEquals(Collections.singleton(topicName), e.unauthorizedTopics()); } }