@Test
	public void testSparkScalaProcessorWithInputType() throws Exception {
		HttpSource source = new HttpSource(shell);
		String streamName =  testName.getMethodName()  + new Random().nextInt();
		FileSink sink = new FileSink().binary(true);
		try {
			String stream = String.format("%s | spark-scala-word-count --inputType=text/plain | " +
					"%s --inputType=text/plain", source, sink);
			createStream(streamName, stream);
			source.ensureReady().postData(TEST_MESSAGE);
			assertThat(sink, eventually(hasContentsThat(equalTo("(foo,3)"))));
		}
		finally {
			streamOps.destroyStream(streamName);
			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