Fetcher { public Map<String, List<PartitionInfo>> getTopicMetadata(MetadataRequest request, long timeout) { if (!request.isAllTopics() && request.topics() .isEmpty()) { return Collections.emptyMap(); } long start = time.milliseconds(); long remaining = timeout; do { RequestFuture<ClientResponse> future = sendMetadataRequest(request); client.poll(future, remaining); if (future.failed() && !future.isRetriable()) { throw future.exception(); } if (future.succeeded()) { MetadataResponse response = new MetadataResponse(future.value() .responseBody()); Cluster cluster = response.cluster(); Set<String> unauthorizedTopics = cluster.unauthorizedTopics(); if (!unauthorizedTopics.isEmpty()) { throw new TopicAuthorizationException(unauthorizedTopics); } boolean shouldRetry = false; Map<String, Errors> errors = response.errors(); if (!errors.isEmpty()) { log.debug("Topic metadata fetch included errors: {}", errors); for (Map.Entry<String, Errors> errorEntry : errors.entrySet()) { String topic = errorEntry.getKey(); Errors error = errorEntry.getValue(); if (error == Errors.INVALID_TOPIC_EXCEPTION) { throw new InvalidTopicException("Topic '" + topic + "' is invalid"); } else if (error == Errors.UNKNOWN_TOPIC_OR_PARTITION) { continue; } else if (error.exception() instanceof RetriableException) { shouldRetry = true; } else { throw new KafkaException("Unexpected error fetching metadata for topic " + topic, error.exception()); } } } if (!shouldRetry) { HashMap<String, List<PartitionInfo>> topicsPartitionInfos = new HashMap<>(); for (String topic : cluster.topics()) topicsPartitionInfos.put(topic, cluster.availablePartitionsForTopic(topic)); return topicsPartitionInfos; } } long elapsed = time.milliseconds() - start; remaining = timeout - elapsed; if (remaining > 0) { long backoff = Math.min(remaining, retryBackoffMs); time.sleep(backoff); remaining -= backoff; } } while (remaining > 0); throw new TimeoutException("Timeout expired while fetching topic metadata"); } 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(expected = InvalidTopicException.class) public void testGetTopicMetadataInvalidTopic() { client.prepareResponse(newMetadataResponse(topicName, Errors.INVALID_TOPIC_EXCEPTION).toStruct()); fetcher.getTopicMetadata(new MetadataRequest(Collections.singletonList(topicName)), 5000L); }
@Test public void testGetTopicMetadataUnknownTopic() { client.prepareResponse(newMetadataResponse(topicName, Errors.UNKNOWN_TOPIC_OR_PARTITION).toStruct()); Map<String, List<PartitionInfo>> topicMetadata = fetcher.getTopicMetadata(new MetadataRequest(Collections.singletonList(topicName)), 5000L); assertNull(topicMetadata.get(topicName)); }
@Test public void testGetTopicMetadataLeaderNotAvailable() { client.prepareResponse(newMetadataResponse(topicName, Errors.LEADER_NOT_AVAILABLE).toStruct()); client.prepareResponse(newMetadataResponse(topicName, Errors.NONE).toStruct()); Map<String, List<PartitionInfo>> topicMetadata = fetcher.getTopicMetadata(new MetadataRequest(Collections.singletonList(topicName)), 5000L); assertTrue(topicMetadata.containsKey(topicName)); }