diff --git a/spring-integration-core/src/main/java/org/springframework/integration/util/IntegrationReactiveUtils.java b/spring-integration-core/src/main/java/org/springframework/integration/util/IntegrationReactiveUtils.java
index 045ff2716b..dfb2bac5a4 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/util/IntegrationReactiveUtils.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/util/IntegrationReactiveUtils.java
@@ -17,6 +17,7 @@
 package org.springframework.integration.util;
 
 import java.time.Duration;
+import java.util.concurrent.locks.LockSupport;
 
 import org.reactivestreams.Publisher;
 
@@ -127,7 +128,11 @@ public final class IntegrationReactiveUtils {
 		return Flux.defer(() -> {
 			Sinks.Many<Message<T>> sink = Sinks.many().multicast().onBackpressureBuffer(1);
 			@SuppressWarnings("unchecked")
-			MessageHandler messageHandler = (message) -> sink.emitNext((Message<T>) message);
+			MessageHandler messageHandler = (message) -> {
+				while (!sink.emitNext((Message<T>) message).hasEmitted()) {
+					LockSupport.parkNanos(10);
+				}
+			};
 			inputChannel.subscribe(messageHandler);
 			return sink.asFlux()
 					.doOnCancel(() -> inputChannel.unsubscribe(messageHandler));
diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/MessageChannelReactiveUtilsTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/MessageChannelReactiveUtilsTests.java
index 697ecdc766..bc0216db05 100644
--- a/spring-integration-core/src/test/java/org/springframework/integration/channel/MessageChannelReactiveUtilsTests.java
+++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/MessageChannelReactiveUtilsTests.java
@@ -21,7 +21,6 @@ import static org.assertj.core.api.Assertions.assertThat;
 import java.time.Duration;
 import java.util.concurrent.atomic.AtomicInteger;
 
-import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 
 import org.springframework.integration.util.IntegrationReactiveUtils;
@@ -71,7 +70,6 @@ class MessageChannelReactiveUtilsTests {
 	}
 
 	@Test
-	@Disabled("Backpressure is not honored")
 	void testOverproducingWithSubscribableChannel() {
 		DirectChannel channel = new DirectChannel();
 
diff --git a/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/client/StompServerIntegrationTests.java b/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/client/StompServerIntegrationTests.java
index 84077a2203..d6905b9197 100644
--- a/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/client/StompServerIntegrationTests.java
+++ b/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/client/StompServerIntegrationTests.java
@@ -22,6 +22,7 @@ import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
 import org.apache.activemq.broker.BrokerService;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 
 import org.springframework.context.ApplicationEvent;
@@ -62,7 +63,7 @@ import org.springframework.util.SocketUtils;
  *
  * @since 4.2
  */
-//@Disabled("Until the fix in reactor-netty-core")
+@Disabled("Until the fix in reactor-netty-core")
 public class StompServerIntegrationTests {
 
 	private static BrokerService activeMQBroker;