public static String findNewLeader(Collection<BrokerReference> replicaBrokers, BrokerReference oldLeader, String topicId, int partitionId) throws Exception {
		for (int i = 0; i < 3; i++) {
			boolean goToSleep = false;
			PartitionMetadata metadata = findLeader(replicaBrokers, replicaBrokers, topicId, partitionId);
			if (metadata == null) {
				goToSleep = true;
			} else if (metadata.leader() == null) {
				goToSleep = true;
			} else if (oldLeader.getHostname().equalsIgnoreCase(metadata.leader().host()) && oldLeader.getPort() == metadata.leader().port() && i == 0) {
				// first time through if the leader hasn't changed give ZooKeeper a second to recover
				// second time, assume the broker did recover before failover, or it was a non-Broker issue
				//
				goToSleep = true;
			} else {
				return metadata.leader().host();
			}
			if (goToSleep) {
				try {
					Thread.sleep(1000);
				} catch (InterruptedException ie) {
				}
			}
		}
		throw new Exception("Unable to find new leader after Broker failure. Exiting");
	}