private void setupRetryTemplate(final String topic) throws Exception {
		RetryTemplate retryTemplate = new RetryTemplate();
		retryTemplate.setRetryPolicy(new SimpleRetryPolicy(5,
				Collections.<Class<? extends Throwable>,Boolean>singletonMap(TopicNotFoundException.class, true)));
		retryTemplate.setBackOffPolicy(new ExponentialBackOffPolicy());
		retryTemplate.execute(new RetryCallback<TopicMetadata, Exception>() {
			@Override
			public TopicMetadata doWithRetry(RetryContext context) throws Exception {
				return AdminUtils.fetchTopicMetadataFromZk(topic, kafkaTestSupport.getZkClient());
			}
		});
	}
Integration tests for Kafka source and sinks.

@author Ilayaperumal Gopinathan
@author Marius Bogoevici
@since 1.1
spring-xd-shell/src/test/java/org/springframework/xd/shell/command/AbstractKafkaSourceSinkTests.java