@Test
	public void testTapSparkProcessor() throws Exception {
		HttpSource source = new HttpSource(shell);
		String streamName =  testName.getMethodName() + new Random().nextInt();
		String tapStreamName =  testName.getMethodName() + new Random().nextInt();
		CounterSink counter = metrics.newCounterSink();
		try {
			String stream = String.format("%s | spark-word-count --enableTap=true | null", source);
			createStream(streamName, stream);
			String tapStream = String.format("tap:stream:%s.spark-word-count > %s --inputType=text/plain", streamName, counter);
			createStream(tapStreamName, tapStream);
			source.ensureReady().postData(TEST_LONG_MESSAGE);
			assertThat(counter, eventually(50, 100, exists()));
			assertThat(counter, eventually(hasValue("4")));
		}
		finally {
			streamOps.destroyStream(streamName);
			streamOps.destroyStream(tapStreamName);
			counter.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