diff --git a/processor-samples/reactive-processor/README.adoc b/processor-samples/reactive-processor/README.adoc
index e62e774..a759538 100644
--- a/processor-samples/reactive-processor/README.adoc
+++ b/processor-samples/reactive-processor/README.adoc
@@ -18,7 +18,7 @@ The following instructions assume that you are running Kafka as a Docker image.
 
 * `./mvnw clean package`
 
-* `java -jar target/reactive-processor-0.0.1-SNAPSHOT.jar`
+* `java -jar target/reactive-processor-0.0.1-SNAPSHOT-kafka.jar`
 
 The main application contains the reactive processor that receives textual data for a duration and aggregates them.
 It then sends the aggregated data through the outbound destination of the processor.
@@ -47,6 +47,6 @@ All the instructions above apply here also, but instead of running the default `
 
 * `./mvnw clean package -P rabbit-binder`
 
-* `java -jar target/reactive-processor-0.0.1-SNAPSHOT.jar`
+* `java -jar target/reactive-processor-0.0.1-SNAPSHOT-rabbit.jar`
 
 Once you are done testing: `docker-compose -f docker-compose-rabbit.yml down`
diff --git a/processor-samples/reactive-processor/src/main/java/reactive/kafka/ReactiveProcessorApplication.java b/processor-samples/reactive-processor/src/main/java/reactive/kafka/ReactiveProcessorApplication.java
index 3ef7a91..dc9ba5e 100644
--- a/processor-samples/reactive-processor/src/main/java/reactive/kafka/ReactiveProcessorApplication.java
+++ b/processor-samples/reactive-processor/src/main/java/reactive/kafka/ReactiveProcessorApplication.java
@@ -20,6 +20,7 @@ import reactor.core.publisher.Flux;
 
 import java.time.Duration;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Function;
 
 @SpringBootApplication
 @EnableBinding(Processor.class)
@@ -29,12 +30,21 @@ public class ReactiveProcessorApplication {
 		SpringApplication.run(ReactiveProcessorApplication.class, args);
 	}
 
-	@StreamListener
-	@Output(Processor.OUTPUT)
-	public Flux<String> aggregate(@Input(Processor.INPUT) Flux<String> inbound) {
-		return inbound.
+//	@StreamListener
+//	@Output(Processor.OUTPUT)
+//	public Flux<String> aggregate(@Input(Processor.INPUT) Flux<String> inbound) {
+//		return inbound.
+//				log()
+//				.window(Duration.ofSeconds(5), Duration.ofSeconds(5))
+//				.flatMap(w -> w.reduce("", (s1,s2)->s1+s2))
+//				.log();
+//	}
+
+	@Bean
+	public Function<Flux<String>, Flux<String>> aggregate() {
+		return inbound -> inbound.
 				log()
-				.window(Duration.ofSeconds(5), Duration.ofSeconds(5))
+				.window(Duration.ofSeconds(30), Duration.ofSeconds(5))
 				.flatMap(w -> w.reduce("", (s1,s2)->s1+s2))
 				.log();
 	}
diff --git a/processor-samples/reactive-processor/src/main/resources/application.yml b/processor-samples/reactive-processor/src/main/resources/application.yml
index e9a9b04..c6e590d 100644
--- a/processor-samples/reactive-processor/src/main/resources/application.yml
+++ b/processor-samples/reactive-processor/src/main/resources/application.yml
@@ -1,6 +1,7 @@
 spring:
   cloud:
     stream:
+      function.definition: aggregate
       bindings:
         output:
           destination: transformed
diff --git a/samples-e2e-tests/src/test/java/sample/acceptance/tests/SampleAcceptanceTests.java b/samples-e2e-tests/src/test/java/sample/acceptance/tests/SampleAcceptanceTests.java
index 5d00981..a62669c 100644
--- a/samples-e2e-tests/src/test/java/sample/acceptance/tests/SampleAcceptanceTests.java
+++ b/samples-e2e-tests/src/test/java/sample/acceptance/tests/SampleAcceptanceTests.java
@@ -241,8 +241,8 @@ public class SampleAcceptanceTests extends AbstractSampleTests {
 		waitForLogEntryInFile("Reactive processor", file,"Started ReactiveProcessorApplication in");
 
 		waitForLogEntryInFile("Reactive processor", file,
-				"Data received: foobarfoobarfoo",
-				"Data received: barfoobarfoobar");
+				"Data received: foo",
+				"Data received: bar");
 	}
 
 	@Test
@@ -256,8 +256,8 @@ public class SampleAcceptanceTests extends AbstractSampleTests {
 		waitForLogEntryInFile("Reactive processor", file,"Started ReactiveProcessorApplication in");
 
 		waitForLogEntryInFile("Reactive processor", file,
-				"Data received: foobarfoobarfoo",
-				"Data received: barfoobarfoobar");
+				"Data received: foo",
+				"Data received: bar");
 	}
 
 	@Test