public static PartitionMetadata findLeader(Collection<BrokerReference> replicaBrokers, Collection<BrokerReference> seedBrokers, String topicId, int partitionId) {
		PartitionMetadata returnMetaData = null;
		loop:
			for (BrokerReference seed : seedBrokers) {
				SimpleConsumer consumer = null;
				try {
					consumer = new SimpleConsumer(seed.getHostname(), seed.getPort(), 100000, 64 * 1024, "leaderLookup");
					List<String> topics = Collections.singletonList(topicId);
					TopicMetadataRequest req = new TopicMetadataRequest(topics);
					TopicMetadataResponse resp = consumer.send(req);

					List<TopicMetadata> metaData = resp.topicsMetadata();
					for (TopicMetadata item : metaData) {
						for (PartitionMetadata part : item.partitionsMetadata()) {
							if (part.partitionId() == partitionId) {
								returnMetaData = part;
								break loop;
							}
						}
					}
				} catch (Exception e) {
					System.out.println("Error communicating with Broker [" + seed + "] to find Leader for [" + topicId
							+ ", " + partitionId + "] Reason: " + e);
				} finally {
					if (consumer != null) consumer.close();
				}
			}
		if (returnMetaData != null) {
			replicaBrokers.clear();
			for (kafka.cluster.Broker replica : returnMetaData.replicas()) {
				replicaBrokers.add(new BrokerReference(replica.host(), replica.port()));
			}
		}
		return returnMetaData;
	}