@SuppressWarnings("rawtypes")
	public void execute(JavaReceiverInputDStream input, Processor processor, final SparkMessageSender sender) {
		JavaDStreamLike output = processor.process(input);
		if (output != null) {
			output.foreachRDD(new Function<JavaRDDLike, Void>() {
				@Override
				public Void call(final JavaRDDLike rdd) {
					rdd.foreachPartition(new VoidFunction<Iterator<?>>() {
						@Override
						public void call(Iterator<?> results) throws Exception {
							if (messageSender == null) {
								messageSender = sender;
								messageSender.start();
							}
							while (results.hasNext()) {
								Object next = results.next();
								Message message = (next instanceof Message) ? (Message) next :
										MessageBuilder.withPayload(next).build();
								messageSender.send(message);
							}
						}
					});
					return null;
				}
			});
			if (messageSender != null) {
				messageSender.stop();
				messageSender = null;
			}
		}
	}
Invokes the process method of a {@link org.springframework.xd.spark.streaming.java.Processor}
and handles the output DStream if present.

@author Ilayaperumal Gopinathan
@author Mark Fisher
@since 1.1
spring-xd-spark-streaming/src/main/java/org/springframework/xd/spark/streaming/java/ModuleExecutor.java