diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java
index 7a4c9600c6..0f6d9aa7ab 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java
@@ -96,7 +96,7 @@ import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.StringUtils;
 
-import reactor.core.publisher.EmitterProcessor;
+import reactor.core.publisher.TopicProcessor;
 import reactor.util.function.Tuple2;
 
 /**
@@ -2728,7 +2728,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 		}
 		else {
 			if (channelForPublisher != null) {
-				Processor<?, ?> processor = EmitterProcessor.create(false);
+				Processor<?, ?> processor = TopicProcessor.share(false);
 				publisher = (Publisher<Message<T>>) processor;
 				Subscriber<Message<?>> subscriber = (Subscriber<Message<?>>) processor;
 				addComponent(new ReactiveConsumer(channelForPublisher, subscriber));
diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/ChannelAdapterParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/ChannelAdapterParserTests.java
index d7d89c2155..c12709d842 100644
--- a/spring-integration-core/src/test/java/org/springframework/integration/config/ChannelAdapterParserTests.java
+++ b/spring-integration-core/src/test/java/org/springframework/integration/config/ChannelAdapterParserTests.java
@@ -55,6 +55,7 @@ import org.springframework.messaging.support.GenericMessage;
 public class ChannelAdapterParserTests {
 
 	private AbstractApplicationContext applicationContext;
+
 	private AbstractApplicationContext applicationContextInner;
 
 
@@ -69,6 +70,7 @@ public class ChannelAdapterParserTests {
 	@After
 	public void tearDown() {
 		this.applicationContext.close();
+		this.applicationContextInner.close();
 	}
 
 
@@ -307,17 +309,21 @@ public class ChannelAdapterParserTests {
 		assertEquals("test", message.getPayload());
 
 		MessageSource<?> testMessageSource = this.applicationContext.getBean("testMessageSource", MessageSource.class);
-		SourcePollingChannelAdapter adapterWithMessageSourceRef = this.applicationContext.getBean("adapterWithMessageSourceRef", SourcePollingChannelAdapter.class);
+		SourcePollingChannelAdapter adapterWithMessageSourceRef =
+				this.applicationContext.getBean("adapterWithMessageSourceRef", SourcePollingChannelAdapter.class);
 		MessageSource<?> source = TestUtils.getPropertyValue(adapterWithMessageSourceRef, "source", MessageSource.class);
 		assertSame(testMessageSource, source);
 	}
 
 	public static class SampleBean {
+
 		private final String message = "hello";
 
 		String getMessage() {
 			return message;
 		}
+
 	}
+
 }
 
diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/reactivestreams/ReactiveStreamsTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/reactivestreams/ReactiveStreamsTests.java
index 8f256c92c5..76a3cf443a 100644
--- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/reactivestreams/ReactiveStreamsTests.java
+++ b/spring-integration-core/src/test/java/org/springframework/integration/dsl/reactivestreams/ReactiveStreamsTests.java
@@ -27,11 +27,9 @@ import java.util.Arrays;
 import java.util.Date;
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.apache.log4j.Level;
@@ -79,7 +77,7 @@ public class ReactiveStreamsTests {
 	private Publisher<Message<Integer>> pollablePublisher;
 
 	@Autowired
-	@Qualifier("reactiveSteamsMessageSource")
+	@Qualifier("reactiveStreamsMessageSource")
 	private Lifecycle messageSource;
 
 	@Autowired
@@ -87,7 +85,7 @@ public class ReactiveStreamsTests {
 	private MessageChannel inputChannel;
 
 	@Test
-	public void testReactiveFlow() throws InterruptedException {
+	public void testReactiveFlow() throws Exception {
 		List<String> results = new ArrayList<>();
 		CountDownLatch latch = new CountDownLatch(6);
 		Flux.from(this.publisher)
@@ -100,10 +98,11 @@ public class ReactiveStreamsTests {
 		assertTrue(latch.await(10, TimeUnit.SECONDS));
 		String[] strings = results.toArray(new String[results.size()]);
 		assertArrayEquals(new String[] { "A", "B", "C", "D", "E", "F" }, strings);
+		this.messageSource.stop();
 	}
 
 	@Test
-	public void testPollableReactiveFlow() throws InterruptedException, TimeoutException, ExecutionException {
+	public void testPollableReactiveFlow() throws Exception {
 		this.inputChannel.send(new GenericMessage<>("1,2,3,4,5"));
 
 		CountDownLatch warmUpLatch = new CountDownLatch(3);
@@ -154,7 +153,7 @@ public class ReactiveStreamsTests {
 					.from(() -> new GenericMessage<>("a,b,c,d,e,f"),
 							e -> e.poller(p -> p.trigger(ctx -> this.invoked.getAndSet(true) ? null : new Date()))
 									.autoStartup(false)
-									.id("reactiveSteamsMessageSource"))
+									.id("reactiveStreamsMessageSource"))
 					.split(String.class, p -> p.split(","))
 					.toReactivePublisher();
 		}