@Override
	protected void obtainResource() throws Exception {
		if (embedded) {
			log.debug("Starting Zookeeper");
			zookeeper = new EmbeddedZookeeper(TestZKUtils.zookeeperConnect());
			log.debug("Started Zookeeper at " + zookeeper.getConnectString());
			try {
				int zkConnectionTimeout = 6000;
				int zkSessionTimeout = 6000;
				zkClient = new ZkClient(getZkConnectString(), zkSessionTimeout, zkConnectionTimeout, ZKStringSerializer$.MODULE$);
			}
			catch (Exception e) {
				zookeeper.shutdown();
				throw e;
			}
			try {
				log.debug("Creating Kafka server");
				Properties brokerConfigProperties = brokerConfig;
				kafkaServer = TestUtils.createServer(new KafkaConfig(brokerConfigProperties), SystemTime$.MODULE$);
				log.debug("Created Kafka server at " + kafkaServer.config().hostName() + ":" + kafkaServer.config().port());
			}
			catch (Exception e) {
				zookeeper.shutdown();
				zkClient.close();
				throw e;
			}
		}
		else {
			this.zkClient = new ZkClient(DEFAULT_ZOOKEEPER_CONNECT, 10000, 10000, ZKStringSerializer$.MODULE$);
			if (ZkUtils.getAllBrokersInCluster(zkClient).size() == 0) {
				throw new RuntimeException("Kafka server not available");
			}
		}
	}
JUnit {@link Rule} that starts an embedded Kafka server (with an associated Zookeeper)

@author Ilayaperumal Gopinathan
@author Marius Bogoevici
@author Gary Russell
@since 1.1
spring-xd-test/src/main/java/org/springframework/xd/test/kafka/KafkaTestSupport.java