@Test
	public void testKafkaSourceAndSinkWithMultiTopics() throws Exception {
		topic2ToUse = "kafka-test-topic-" + random.nextInt();
		AdminUtils.createTopic(kafkaTestSupport.getZkClient(), topic2ToUse, 1, 1, new Properties());
		setupRetryTemplate(topic2ToUse);
		final String stringToPost = "Hi there!";
		// create stream with kafka sink
		final HttpSource httpSource1 = newHttpSource();
		final HttpSource httpSource2 = newHttpSource();
		stream().create(generateStreamName(), "%s | kafka --topic='%s' --brokerList='%s'",
				httpSource1, topic1ToUse, kafkaTestSupport.getBrokerAddress());
		stream().create(generateStreamName(), "%s | kafka --topic='%s' --brokerList='%s'",
				httpSource2, topic2ToUse, kafkaTestSupport.getBrokerAddress());
		// create stream with kafka source
		final CounterSink counter = metrics().newCounterSink();
		stream().create(generateStreamName(), "kafka --topics='%s,%s' --zkconnect=%s --outputType=text/plain | " +
				"filter --expression=payload.toString().contains('%s') | %s",
				topic1ToUse, topic2ToUse, kafkaTestSupport.getZkConnectString(), stringToPost, counter );
		httpSource1.ensureReady().postData(stringToPost);
		httpSource2.ensureReady().postData(stringToPost);
		assertThat(counter, eventually(exists()));
		assertThat(counter, eventually(hasValue("2")));
		AdminUtils.deleteTopic(kafkaTestSupport.getZkClient(), topic2ToUse);
	}
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