@Override
				public Void doWithRetry(RetryContext context) throws Exception {
					TopicMetadata topicMetadata = AdminUtils.fetchTopicMetadataFromZk(topic, zkClient);
					if (topicMetadata.errorCode() != ErrorMapping.NoError() || !topic.equals(topicMetadata.topic())) {
						// downcast to Exception because that's what the error throws
						throw (Exception) ErrorMapping.exceptionFor(topicMetadata.errorCode());
					}
					List<PartitionMetadata> partitionMetadatas = new kafka.javaapi.TopicMetadata(topicMetadata).partitionsMetadata();
					for (PartitionMetadata partitionMetadata : partitionMetadatas) {
						if (partitionMetadata.errorCode() != ErrorMapping.NoError()) {
							throw (Error) ErrorMapping.exceptionFor(partitionMetadata.errorCode());
						}
					}
					return null;
				}
Ensures that the host and port of the zookeeper connection are available and
that the topic has been created.
@param fixtureName  The name of the fixture calling htis utility method, used for error reporting
@param zkConnect The Zookeeper connection string
@param topic The name of the topic to create
spring-xd-test-fixtures/src/main/java/org/springframework/xd/test/fixtures/util/KafkaUtils.java