@Test
	public void testSparkTapStream() throws Exception {
		HttpSource source = new HttpSource(shell);
		String streamName =  testName.getMethodName() + new Random().nextInt();
		String tapStreamName =  testName.getMethodName() + new Random().nextInt();
		FileSink sink = new FileSink().binary(true);
		try {
			String stream = String.format("%s | counter", source);
			createStream(streamName, stream);
			String tapStream = String.format("tap:stream:%s > spark-word-count | %s --inputType=text/plain",
					streamName, sink);
			createStream(tapStreamName, tapStream);
			source.ensureReady().postData(TEST_LONG_MESSAGE);
			assertThat(sink, eventually(hasContentsThat(containsString("(foo,6)"))));
			assertThat(sink, eventually(hasContentsThat(containsString("(bar,5)"))));
			assertThat(sink, eventually(hasContentsThat(containsString("(test1,2)"))));
			assertThat(sink, eventually(hasContentsThat(containsString("(test2,1)"))));
		}
		finally {
			streamOps.destroyStream(streamName);
			streamOps.destroyStream(tapStreamName);
			sink.cleanup();
		}
	}
Abstract Spark streaming test class which can be extended to run against multiple XD transport.

@author Ilayaperumal Gopinathan
spring-xd-spark-streaming-tests/src/test/java/org/springframework/xd/spark/streaming/AbstractSparkStreamingTests.java