docs: private static Map<Node, List<KafkaTopicPartitionState<TopicAndPartition>>> findLeaderForPartitions(
  			List<KafkaTopicPartitionState<TopicAndPartition>> partitionsToAssign,
  			Properties kafkaProperties) throws Exception {
  		if (partitionsToAssign.isEmpty()) {
  			throw new IllegalArgumentException("Leader request for empty partitions list");
  		}
  
  		LOG.info("Refreshing leader information for partitions {}", partitionsToAssign);
  
  		// this request is based on the topic names
  		PartitionInfoFetcher infoFetcher = new PartitionInfoFetcher(getTopics(partitionsToAssign), kafkaProperties);
  		infoFetcher.start();
  
  		// NOTE: The kafka client apparently locks itself up sometimes
  		// when it is interrupted, so we run it only in a separate thread.
  		// since it sometimes refuses to shut down, we resort to the admittedly harsh
  		// means of killing the thread after a timeout.
  		KillerWatchDog watchDog = new KillerWatchDog(infoFetcher, 60000);
  		watchDog.start();
  
  		// this list contains ALL partitions of the requested topics
  		List<KafkaTopicPartitionLeader> topicPartitionWithLeaderList = infoFetcher.getPartitions();
  
  		// copy list to track unassigned partitions
  		List<KafkaTopicPartitionState<TopicAndPartition>> unassignedPartitions = new ArrayList<>(partitionsToAssign);
  
  		// final mapping from leader -> list(fetchPartition)
  		Map<Node, List<KafkaTopicPartitionState<TopicAndPartition>>> leaderToPartitions = new HashMap<>();
  
  		for (KafkaTopicPartitionLeader partitionLeader: topicPartitionWithLeaderList) {
  			if (unassignedPartitions.size() == 0) {
  				// we are done: all partitions are assigned
  				break;
  			}
  
  			Iterator<KafkaTopicPartitionState<TopicAndPartition>> unassignedPartitionsIterator = unassignedPartitions.iterator();
  			while (unassignedPartitionsIterator.hasNext()) {
  				KafkaTopicPartitionState<TopicAndPartition> unassignedPartition = unassignedPartitionsIterator.next();
  
  				if (unassignedPartition.getKafkaTopicPartition().equals(partitionLeader.getTopicPartition())) {
  					// we found the leader for one of the fetch partitions
  					Node leader = partitionLeader.getLeader();
  
  					List<KafkaTopicPartitionState<TopicAndPartition>> partitionsOfLeader = leaderToPartitions.get(leader);
  					if (partitionsOfLeader == null) {
  						partitionsOfLeader = new ArrayList<>();
  						leaderToPartitions.put(leader, partitionsOfLeader);
  					}
  					partitionsOfLeader.add(unassignedPartition);
  					unassignedPartitionsIterator.remove(); // partition has been assigned
  					break;
  				}
  			}
  		}
  
  		if (unassignedPartitions.size() > 0) {
  			throw new RuntimeException("Unable to find a leader for partitions: " + unassignedPartitions);
  		}
  
  		LOG.debug("Partitions with assigned leaders {}", leaderToPartitions);
  
  		return leaderToPartitions;
  	}