@Override
    public List<PartitionInfo> partitionsFor(String topic, Duration timeout) {
        acquireAndEnsureOpen();
        try {
            Cluster cluster = this.metadata.fetch();
            List<PartitionInfo> parts = cluster.partitionsForTopic(topic);
            if (!parts.isEmpty())
                return parts;

            Timer timer = time.timer(timeout);
            List<PartitionInfo> topicMetadata = topicMetadataFetcher.getTopicMetadata(topic, metadata.allowAutoTopicCreation(), timer);
            return topicMetadata != null ? topicMetadata : Collections.emptyList();
        } finally {
            release();
        }
    }
@throws KafkaException if the rebalance callback throws exception
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ClassicKafkaConsumer.java