public static void ensureReady(String fixtureName, String zkConnect, final String topic) {
        String[] addressArray = StringUtils.commaDelimitedListToStringArray(zkConnect);
        for (String address : addressArray) {
            String[] zkAddressArray = StringUtils.delimitedListToStringArray(address, ":");
            Assert.isTrue(zkAddressArray.length == 2,
                    "zkConnect data was not properly formatted");
            String host = zkAddressArray[0];
            int port = Integer.valueOf(zkAddressArray[1]);
            AvailableSocketPorts.ensureReady(fixtureName, host, port, 2000);
        }
        final ZkClient zkClient = new ZkClient(zkConnect, 6000, 6000, ZKStringSerializer$.MODULE$);
        AdminUtils.createTopic(zkClient, topic, 1, 1, new Properties());

		RetryTemplate retryTemplate = new RetryTemplate();

		CompositeRetryPolicy policy = new CompositeRetryPolicy();
		TimeoutRetryPolicy timeoutRetryPolicy = new TimeoutRetryPolicy();
		timeoutRetryPolicy.setTimeout(METADATA_VERIFICATION_TIMEOUT);
		SimpleRetryPolicy simpleRetryPolicy = new SimpleRetryPolicy();
		simpleRetryPolicy.setMaxAttempts(METADATA_VERIFICATION_RETRY_ATTEMPTS);
		policy.setPolicies(new RetryPolicy[]{timeoutRetryPolicy, simpleRetryPolicy});
		retryTemplate.setRetryPolicy(policy);

		ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
		backOffPolicy.setInitialInterval(METADATA_VERIFICATION_RETRY_INITIAL_INTERVAL);
		backOffPolicy.setMultiplier(METADATA_VERIFICATION_RETRY_BACKOFF_MULTIPLIER);
		backOffPolicy.setMaxInterval(METADATA_VERIFICATION_MAX_INTERVAL);
		retryTemplate.setBackOffPolicy(backOffPolicy);

		try {
			retryTemplate.execute(new RetryCallback<Void, Exception>() {
				@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;
				}
			});
		}
		catch (Exception e) {
			throw new IllegalStateException("Unable to create topic for kafka", e);
		}
	}
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