Clean up some schedulers and contexts in tests

* Migrate affected test to JUnit 5

All the work is done with an attempt to figure out an OOM errors on CI

spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2020 the original author or authors.
+ * Copyright 2015-2021 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -32,6 +32,7 @@ import reactor.core.Disposables;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.Sinks;
+import reactor.core.scheduler.Scheduler;
 import reactor.core.scheduler.Schedulers;
 
 /**
@@ -47,6 +48,8 @@ import reactor.core.scheduler.Schedulers;
 public class FluxMessageChannel extends AbstractMessageChannel
 		implements Publisher<Message<?>>, ReactiveStreamsSubscribableChannel {
 
+	private final Scheduler scheduler = Schedulers.boundedElastic();
+
 	private final Sinks.Many<Message<?>> sink = Sinks.many().multicast().onBackpressureBuffer(1, false);
 
 	private final Sinks.Many<Boolean> subscribedSignal = Sinks.many().replay().limit(1);
@@ -114,7 +117,7 @@ public class FluxMessageChannel extends AbstractMessageChannel
 		this.upstreamSubscriptions.add(
 				Flux.from(publisher)
 						.delaySubscription(this.subscribedSignal.asFlux().filter(Boolean::booleanValue).next())
-						.publishOn(Schedulers.boundedElastic())
+						.publishOn(this.scheduler)
 						.doOnNext((message) -> {
 							try {
 								if (!send(message)) {
@@ -135,6 +138,7 @@ public class FluxMessageChannel extends AbstractMessageChannel
 		this.upstreamSubscriptions.dispose();
 		this.subscribedSignal.emitComplete(Sinks.EmitFailureHandler.FAIL_FAST);
 		this.sink.emitComplete(Sinks.EmitFailureHandler.FAIL_FAST);
+		this.scheduler.dispose();
 		super.destroy();
 	}
 

spring-integration-core/src/test/java/org/springframework/integration/aop/PublisherAnnotationAdvisorTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2021 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -23,8 +23,9 @@ import java.lang.annotation.Retention;
 import java.lang.annotation.RetentionPolicy;
 import java.lang.annotation.Target;
 
-import org.junit.Before;
-import org.junit.Test;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
 
 import org.springframework.aop.framework.ProxyFactory;
 import org.springframework.context.support.StaticApplicationContext;
@@ -36,6 +37,7 @@ import org.springframework.messaging.handler.annotation.Payload;
 /**
  * @author Mark Fisher
  * @author Jeff Maxwell
+ * @author Artem Bilan
  *
  * @since 2.0
  */
@@ -43,12 +45,17 @@ public class PublisherAnnotationAdvisorTests {
 
 	private final StaticApplicationContext context = new StaticApplicationContext();
 
-	@Before
+	@BeforeEach
 	public void setup() {
 		context.registerSingleton("testChannel", QueueChannel.class);
 		context.registerSingleton("testMetaChannel", QueueChannel.class);
 	}
 
+	@AfterEach
+	void tearDown() {
+		this.context.close();
+	}
+
 	@Test
 	public void annotationAtMethodLevelOnVoidReturnWithParamAnnotation() {
 		PublisherAnnotationAdvisor advisor = new PublisherAnnotationAdvisor();

spring-integration-core/src/test/java/org/springframework/integration/channel/DispatchingChannelErrorHandlingTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2021 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -17,11 +17,12 @@
 package org.springframework.integration.channel;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
 
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
 
 import org.springframework.context.support.StaticApplicationContext;
 import org.springframework.core.task.SimpleAsyncTaskExecutor;
@@ -35,21 +36,23 @@ import org.springframework.messaging.MessagingException;
 
 /**
  * @author Mark Fisher
+ * @author Artem Bilan
+ *
  * @since 1.0.3
  */
 public class DispatchingChannelErrorHandlingTests {
 
 	private final CountDownLatch latch = new CountDownLatch(1);
 
-
-	@Test(expected = MessageDeliveryException.class)
+	@Test
 	public void handlerThrowsExceptionPublishSubscribeWithoutExecutor() {
 		PublishSubscribeChannel channel = new PublishSubscribeChannel();
 		channel.subscribe(message -> {
 			throw new UnsupportedOperationException("intentional test failure");
 		});
 		Message<?> message = MessageBuilder.withPayload("test").build();
-		channel.send(message);
+		assertThatExceptionOfType(MessageDeliveryException.class)
+				.isThrownBy(() -> channel.send(message));
 	}
 
 	@Test
@@ -72,13 +75,14 @@ public class DispatchingChannelErrorHandlingTests {
 		});
 		Message<?> message = MessageBuilder.withPayload("test").build();
 		channel.send(message);
-		this.waitForLatch(10000);
+		waitForLatch(10000);
 		Message<?> errorMessage = resultHandler.lastMessage;
 		assertThat(errorMessage.getPayload().getClass()).isEqualTo(MessagingException.class);
 		MessagingException exceptionPayload = (MessagingException) errorMessage.getPayload();
 		assertThat(exceptionPayload.getCause().getClass()).isEqualTo(UnsupportedOperationException.class);
 		assertThat(exceptionPayload.getFailedMessage()).isSameAs(message);
 		assertThat(resultHandler.lastThread).isNotSameAs(Thread.currentThread());
+		context.close();
 	}
 
 	@Test
@@ -108,6 +112,7 @@ public class DispatchingChannelErrorHandlingTests {
 		assertThat(exceptionPayload.getCause().getClass()).isEqualTo(UnsupportedOperationException.class);
 		assertThat(exceptionPayload.getFailedMessage()).isSameAs(message);
 		assertThat(resultHandler.lastThread).isNotSameAs(Thread.currentThread());
+		context.close();
 	}
 
 
@@ -136,8 +141,8 @@ public class DispatchingChannelErrorHandlingTests {
 			this.lastThread = Thread.currentThread();
 			latch.countDown();
 		}
-	}
 
+	}
 
 	@SuppressWarnings("serial")
 	private static class TestTimedOutException extends RuntimeException {
@@ -145,6 +150,7 @@ public class DispatchingChannelErrorHandlingTests {
 		TestTimedOutException() {
 			super("timed out while waiting for latch");
 		}
+
 	}
 
 }

spring-integration-core/src/test/java/org/springframework/integration/channel/MessageChannelReactiveUtilsTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019-2020 the original author or authors.
+ * Copyright 2019-2021 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -23,6 +23,7 @@ import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.Test;
 
 import org.springframework.integration.util.IntegrationReactiveUtils;
@@ -31,6 +32,7 @@ import org.springframework.messaging.support.GenericMessage;
 import reactor.core.Disposable;
 import reactor.core.Disposables;
 import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Scheduler;
 import reactor.core.scheduler.Schedulers;
 import reactor.test.StepVerifier;
 import reactor.util.concurrent.Queues;
@@ -43,6 +45,13 @@ import reactor.util.concurrent.Queues;
  */
 class MessageChannelReactiveUtilsTests {
 
+	private static final Scheduler SCHEDULER = Schedulers.boundedElastic();
+
+	@AfterAll
+	static void tearDown() {
+		SCHEDULER.dispose();
+	}
+
 	@Test
 	void testBackpressureWithSubscribableChannel() {
 		Disposable.Composite compositeDisposable = Disposables.composite();
@@ -53,7 +62,7 @@ class MessageChannelReactiveUtilsTests {
 					.expectSubscription()
 					.then(() -> {
 						compositeDisposable.add(
-								Schedulers.boundedElastic().schedule(() -> {
+								SCHEDULER.schedule(() -> {
 									while (true) {
 										if (channel.getSubscriberCount() > 0) {
 											channel.send(new GenericMessage<>("foo"));
@@ -84,7 +93,7 @@ class MessageChannelReactiveUtilsTests {
 					.expectSubscription()
 					.then(() ->
 							compositeDisposable.add(
-									Schedulers.boundedElastic().schedule(() -> {
+									SCHEDULER.schedule(() -> {
 										while (true) {
 											if (channel.getSubscriberCount() > 0) {
 												channel.send(new GenericMessage<>("foo"));

spring-integration-core/src/test/java/org/springframework/integration/channel/MixedDispatcherConfigurationScenarioTests-context.xml

@@ -18,6 +18,7 @@
 	<channel id="noLoadBalancerNoFailover" >
 		<dispatcher load-balancer="none" failover="${failover}"/>
 	</channel>
+
 	<channel id="noLoadBalancerNoFailoverExecutor" >
 		<dispatcher load-balancer="none" failover="false" task-executor="taskExecutor"/>
 	</channel>
@@ -25,6 +26,7 @@
 	<channel id="loadBalancerNoFailover">
 		<dispatcher load-balancer="round-robin" failover="false"/>
 	</channel>
+
 	<channel id="loadBalancerNoFailoverExecutor">
 		<dispatcher load-balancer="round-robin" failover="false" task-executor="taskExecutor"/>
 	</channel>
@@ -32,11 +34,13 @@
 	<channel id="noLoadBalancerFailover">
 		<dispatcher load-balancer="none" failover="true"/>
 	</channel>
+
 	<channel id="noLoadBalancerFailoverExecutor">
 		<dispatcher load-balancer="none" failover="true" task-executor="taskExecutor"/>
 	</channel>
 
-	<beans:bean id="taskExecutor" class="java.util.concurrent.Executors" factory-method="newCachedThreadPool"/>
+	<beans:bean id="taskExecutor" class="java.util.concurrent.Executors" factory-method="newCachedThreadPool"
+				destroy-method="shutdown"/>
 
 	<!-- send async errors to null channel to avoid chatter -->
 	<bridge input-channel="errorChannel" output-channel="nullChannel"/>

spring-integration-core/src/test/java/org/springframework/integration/channel/MixedDispatcherConfigurationScenarioTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2021 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -21,88 +21,72 @@ import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.inOrder;
+import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Test;
-import org.junit.runner.RunWith;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
 import org.mockito.InOrder;
-import org.mockito.Mock;
 import org.mockito.Mockito;
-import org.mockito.junit.MockitoJUnitRunner;
 
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.context.ConfigurableApplicationContext;
-import org.springframework.context.support.ClassPathXmlApplicationContext;
 import org.springframework.integration.MessageRejectedException;
 import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy;
 import org.springframework.integration.dispatcher.UnicastingDispatcher;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.messaging.support.GenericMessage;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 
 /**
  * @author Oleg Zhurakousky
  * @author Gary Russell
+ * @author Artem Bilan
  */
-@RunWith(MockitoJUnitRunner.class)
+@SpringJUnitConfig
+@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_EACH_TEST_METHOD)
 public class MixedDispatcherConfigurationScenarioTests {
 
 	private static final int TOTAL_EXECUTIONS = 40;
 
-	private ExecutorService executor;
+	private final CountDownLatch allDone = new CountDownLatch(TOTAL_EXECUTIONS);
 
-	private CountDownLatch allDone;
+	private final CountDownLatch start = new CountDownLatch(1);
 
-	private CountDownLatch start;
-
-	private AtomicBoolean failed;
-
-	@Mock
-	private List<Exception> exceptionRegistry;
-
-	private ConfigurableApplicationContext ac;
-
-	@Mock
-	private MessageHandler handlerA;
-
-	@Mock
-	private MessageHandler handlerB;
-
-	@Mock
-	private MessageHandler handlerC;
-
-	private final Message<?> message = new GenericMessage<String>("test");
+	private final AtomicBoolean failed = new AtomicBoolean(false);
 
+	private final Message<?> message = new GenericMessage<>("test");
 
 	@SuppressWarnings("unchecked")
-	@Before
-	public void initialize() throws Exception {
-		Mockito.reset(exceptionRegistry);
-		Mockito.reset(handlerA);
-		Mockito.reset(handlerB);
-		Mockito.reset(handlerC);
+	private final List<Exception> exceptionRegistry = mock(List.class);
 
-		ac = new ClassPathXmlApplicationContext("MixedDispatcherConfigurationScenarioTests-context.xml",
-				MixedDispatcherConfigurationScenarioTests.class);
-		executor = ac.getBean("taskExecutor", ExecutorService.class);
-		allDone = new CountDownLatch(TOTAL_EXECUTIONS);
-		start = new CountDownLatch(1);
-		failed = new AtomicBoolean(false);
-	}
+	private final MessageHandler handlerA = mock(MessageHandler.class);
 
-	@After
-	public void tearDown() {
-		this.executor.shutdownNow();
-		this.ac.close();
+	private final MessageHandler handlerB = mock(MessageHandler.class);
+
+	private final MessageHandler handlerC = mock(MessageHandler.class);
+
+	@Autowired
+	@Qualifier("taskExecutor")
+	private Executor executor;
+
+	@Autowired
+	private ConfigurableApplicationContext ac;
+
+	@BeforeEach
+	public void initialize() {
+		Mockito.reset(this.exceptionRegistry, this.handlerA, this.handlerB, this.handlerC);
 	}
 
 	@Test
@@ -159,9 +143,6 @@ public class MixedDispatcherConfigurationScenarioTests {
 		start.countDown();
 		assertThat(allDone.await(10, TimeUnit.SECONDS)).isTrue();
 
-		executor.shutdown();
-		executor.awaitTermination(10, TimeUnit.SECONDS);
-
 		assertThat(failed.get()).as("not all messages were accepted").isTrue();
 		verify(handlerA, times(TOTAL_EXECUTIONS)).handleMessage(message);
 		verify(handlerB, times(0)).handleMessage(message);
@@ -169,8 +150,7 @@ public class MixedDispatcherConfigurationScenarioTests {
 	}
 
 	@Test
-	public void noFailoverNoLoadBalancingWithExecutorConcurrent()
-			throws Exception {
+	public void noFailoverNoLoadBalancingWithExecutorConcurrent() throws Exception {
 		final ExecutorChannel channel = (ExecutorChannel) ac.getBean("noLoadBalancerNoFailoverExecutor");
 		UnicastingDispatcher dispatcher = channel.getDispatcher();
 		dispatcher.addHandler(handlerA);
@@ -200,9 +180,6 @@ public class MixedDispatcherConfigurationScenarioTests {
 		start.countDown();
 		assertThat(allDone.await(10, TimeUnit.SECONDS)).isTrue();
 
-		executor.shutdown();
-		executor.awaitTermination(10, TimeUnit.SECONDS);
-
 		assertThat(failed.get()).as("not all messages were accepted").isTrue();
 		verify(handlerA, times(TOTAL_EXECUTIONS)).handleMessage(message);
 		verify(handlerB, times(0)).handleMessage(message);
@@ -281,9 +258,6 @@ public class MixedDispatcherConfigurationScenarioTests {
 		start.countDown();
 		assertThat(allDone.await(10, TimeUnit.SECONDS)).isTrue();
 
-		executor.shutdown();
-		executor.awaitTermination(10, TimeUnit.SECONDS);
-
 		assertThat(failed.get()).as("not all messages were accepted").isTrue();
 		verify(handlerA, times(14)).handleMessage(message);
 		verify(handlerB, times(13)).handleMessage(message);
@@ -334,9 +308,6 @@ public class MixedDispatcherConfigurationScenarioTests {
 		start.countDown();
 		assertThat(allDone.await(10, TimeUnit.SECONDS)).isTrue();
 
-		executor.shutdown();
-		executor.awaitTermination(10, TimeUnit.SECONDS);
-
 		assertThat(failed.get()).as("not all messages were accepted").isTrue();
 		verify(handlerA, times(14)).handleMessage(message);
 		verify(handlerB, times(13)).handleMessage(message);
@@ -346,8 +317,7 @@ public class MixedDispatcherConfigurationScenarioTests {
 
 	@Test
 	public void failoverNoLoadBalancing() {
-		DirectChannel channel = (DirectChannel) ac
-				.getBean("noLoadBalancerFailover");
+		DirectChannel channel = ac.getBean("noLoadBalancerFailover", DirectChannel.class);
 		doThrow(new MessageRejectedException(message, null)).when(handlerA)
 				.handleMessage(message);
 		UnicastingDispatcher dispatcher = channel.getDispatcher();
@@ -376,10 +346,8 @@ public class MixedDispatcherConfigurationScenarioTests {
 	}
 
 	@Test
-	public void failoverNoLoadBalancingConcurrent()
-			throws Exception {
-		final DirectChannel channel = (DirectChannel) ac
-				.getBean("noLoadBalancerFailover");
+	public void failoverNoLoadBalancingConcurrent() throws Exception {
+		final DirectChannel channel = ac.getBean("noLoadBalancerFailover", DirectChannel.class);
 		doThrow(new MessageRejectedException(message, null)).when(handlerA).handleMessage(message);
 		UnicastingDispatcher dispatcher = channel.getDispatcher();
 		dispatcher.addHandler(handlerA);
@@ -415,9 +383,6 @@ public class MixedDispatcherConfigurationScenarioTests {
 		start.countDown();
 		assertThat(allDone.await(10, TimeUnit.SECONDS)).isTrue();
 
-		executor.shutdown();
-		executor.awaitTermination(10, TimeUnit.SECONDS);
-
 		assertThat(failed.get()).as("not all messages were accepted").isFalse();
 		verify(handlerA, times(TOTAL_EXECUTIONS)).handleMessage(message);
 		verify(handlerB, times(TOTAL_EXECUTIONS)).handleMessage(message);
@@ -459,9 +424,6 @@ public class MixedDispatcherConfigurationScenarioTests {
 		start.countDown();
 		assertThat(allDone.await(10, TimeUnit.SECONDS)).isTrue();
 
-		executor.shutdown();
-		executor.awaitTermination(10, TimeUnit.SECONDS);
-
 		verify(handlerA, times(TOTAL_EXECUTIONS)).handleMessage(message);
 		verify(handlerB, times(TOTAL_EXECUTIONS)).handleMessage(message);
 		verify(handlerC, never()).handleMessage(message);

spring-integration-core/src/test/java/org/springframework/integration/channel/QueueChannelTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2021 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -29,9 +29,7 @@ import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
+import org.junit.jupiter.api.Test;
 
 import org.springframework.integration.selector.UnexpiredMessageSelector;
 import org.springframework.integration.support.MessageBuilder;
@@ -286,121 +284,4 @@ public class QueueChannelTests {
 		assertThat(channel.send(new GenericMessage<>("roomAvailable"), 0)).isTrue();
 	}
 
-	@Rule
-	public final TemporaryFolder tempFolder = new TemporaryFolder();
-
-	/*TODO: No Reactor Chronicle artifact
-	@Test
-	public void testReactorPersistentQueue() throws InterruptedException, IOException {
-		final AtomicBoolean messageReceived = new AtomicBoolean(false);
-		final CountDownLatch latch = new CountDownLatch(1);
-		PersistentQueue<Message<?>> queue = new PersistentQueueSpec<Message<?>>()
-				.codec(new JavaSerializationCodec<Message<?>>())
-				.basePath(this.tempFolder.getRoot().getAbsolutePath())
-				.get();
-
-		final QueueChannel channel = new QueueChannel(queue);
-		new Thread(new Runnable() {
-			@Override
-			public void run() {
-				Message<?> message = channel.receive();
-				if (message != null) {
-					messageReceived.set(true);
-					latch.countDown();
-				}
-			}
-		}).start();
-		assertFalse(messageReceived.get());
-		channel.send(new GenericMessage<String>("testing"));
-		latch.await(1000, TimeUnit.MILLISECONDS);
-		assertTrue(messageReceived.get());
-
-		final CountDownLatch latch1 = new CountDownLatch(2);
-
-		Thread thread = new Thread(new Runnable() {
-			@Override
-			public void run() {
-				while (true) {
-					Message<?> message = channel.receive(100);
-					if (message != null) {
-						latch1.countDown();
-						if (latch1.getCount() == 0) {
-							break;
-						}
-					}
-				}
-			}
-		});
-		thread.start();
-
-		Thread.sleep(200);
-		channel.send(new GenericMessage<String>("testing"));
-		channel.send(new GenericMessage<String>("testing"));
-		assertTrue(latch1.await(1000, TimeUnit.MILLISECONDS));
-
-		final AtomicBoolean receiveInterrupted = new AtomicBoolean(false);
-		final CountDownLatch latch2 = new CountDownLatch(1);
-		Thread t = new Thread(new Runnable() {
-			@Override
-			public void run() {
-				Message<?> message = channel.receive(10000);
-				receiveInterrupted.set(true);
-				assertTrue(message == null);
-				latch2.countDown();
-			}
-		});
-		t.start();
-		assertFalse(receiveInterrupted.get());
-		t.interrupt();
-		latch2.await();
-		assertTrue(receiveInterrupted.get());
-
-		receiveInterrupted.set(false);
-		final CountDownLatch latch3 = new CountDownLatch(1);
-		t = new Thread(new Runnable() {
-			@Override
-			public void run() {
-				Message<?> message = channel.receive();
-				receiveInterrupted.set(true);
-				assertTrue(message == null);
-				latch3.countDown();
-			}
-		});
-		t.start();
-		assertFalse(receiveInterrupted.get());
-		t.interrupt();
-		latch3.await();
-		assertTrue(receiveInterrupted.get());
-
-		GenericMessage<String> message1 = new GenericMessage<String>("test1");
-		GenericMessage<String> message2 = new GenericMessage<String>("test2");
-		assertTrue(channel.send(message1));
-		assertTrue(channel.send(message2));
-		List<Message<?>> clearedMessages = channel.clear();
-		assertNotNull(clearedMessages);
-		assertEquals(2, clearedMessages.size());
-
-		clearedMessages = channel.clear();
-		assertNotNull(clearedMessages);
-		assertEquals(0, clearedMessages.size());
-
-		// Test on artificial infinite wait
-		// channel.receive();
-
-		// Distributed scenario
-		final CountDownLatch latch4 = new CountDownLatch(1);
-		new Thread(new Runnable() {
-			@Override
-			public void run() {
-				Message<?> message = channel.receive();
-				if (message != null) {
-					latch4.countDown();
-				}
-			}
-		}).start();
-		queue.add(new GenericMessage<String>("foo"));
-		assertTrue(latch4.await(1000, TimeUnit.MILLISECONDS));
-	}
-*/
-
 }

spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveStreamsConsumerTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2020 the original author or authors.
+ * Copyright 2016-2021 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -110,6 +110,7 @@ public class ReactiveStreamsConsumerTests {
 		assertThat(result).containsExactly(testMessage, testMessage2);
 
 		reactiveConsumer.stop();
+		testChannel.destroy();
 	}
 
 
@@ -291,6 +292,7 @@ public class ReactiveStreamsConsumerTests {
 		assertThat(result).containsExactly(testMessage, testMessage2, testMessage2);
 
 		endpointFactoryBean.stop();
+		testChannel.destroy();
 	}
 
 	@Test
@@ -331,6 +333,7 @@ public class ReactiveStreamsConsumerTests {
 				.verify(Duration.ofSeconds(10));
 
 		reactiveConsumer.stop();
+		testChannel.destroy();
 	}
 
 }