docs: public List<KafkaTopicPartitionLeader> getPartitionLeadersForTopics(List<String> topics) {
  		List<KafkaTopicPartitionLeader> partitions = new LinkedList<>();
  
  		retryLoop: for (int retry = 0; retry < numRetries; retry++) {
  			brokersLoop: for (int arrIdx = 0; arrIdx < seedBrokerAddresses.length; arrIdx++) {
  				LOG.info("Trying to get topic metadata from broker {} in try {}/{}", seedBrokerAddresses[currentContactSeedBrokerIndex], retry, numRetries);
  
  				try {
  					// clear in case we have an incomplete list from previous tries
  					partitions.clear();
  
  					for (TopicMetadata item : consumer.send(new TopicMetadataRequest(topics)).topicsMetadata()) {
  						if (item.errorCode() != ErrorMapping.NoError()) {
  							// warn and try more brokers
  							LOG.warn("Error while getting metadata from broker {} to find partitions for {}. Error: {}.",
  								seedBrokerAddresses[currentContactSeedBrokerIndex], topics.toString(), ErrorMapping.exceptionFor(item.errorCode()).getMessage());
  
  							useNextAddressAsNewContactSeedBroker();
  							continue brokersLoop;
  						}
  
  						if (!topics.contains(item.topic())) {
  							LOG.warn("Received metadata from topic " + item.topic() + " even though it was not requested. Skipping ...");
  
  							useNextAddressAsNewContactSeedBroker();
  							continue brokersLoop;
  						}
  
  						for (PartitionMetadata part : item.partitionsMetadata()) {
  							Node leader = brokerToNode(part.leader());
  							KafkaTopicPartition ktp = new KafkaTopicPartition(item.topic(), part.partitionId());
  							KafkaTopicPartitionLeader pInfo = new KafkaTopicPartitionLeader(ktp, leader);
  							partitions.add(pInfo);
  						}
  					}
  					break retryLoop; // leave the loop through the brokers
  				}
  				catch (Exception e) {
  					//validates seed brokers in case of a ClosedChannelException
  					validateSeedBrokers(seedBrokerAddresses, e);
  					LOG.warn("Error communicating with broker {} to find partitions for {}. {} Message: {}",
  						seedBrokerAddresses[currentContactSeedBrokerIndex], topics, e.getClass().getName(), e.getMessage());
  					LOG.debug("Detailed trace", e);
  
  					// we sleep a bit. Retrying immediately doesn't make sense in cases where Kafka is reorganizing the leader metadata
  					try {
  						Thread.sleep(500);
  					} catch (InterruptedException e1) {
  						// sleep shorter.
  					}
  
  					useNextAddressAsNewContactSeedBroker();
  				}
  			} // brokers loop
  		} // retries loop
  
  		return partitions;
  	}