From ec2cfeb9b82a77c166b807c86d6a0f776147ca4b Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 13 Aug 2024 15:57:25 -0400 Subject: [PATCH] Fix tests for `executorService.shutdown()` Even if `Executors.newSingleThreadExecutor()` returns a `FinalizableDelegatedExecutorService`, an instance is kept in the memory until JVM exists. That may lead to memory leak since we have a lot of threads in memory. (cherry picked from commit fdac8f16342d38136535a61f7a92b9210151ab47) --- .../FluxAggregatorMessageHandlerTests.java | 30 +++++++------- .../config/ThreadLocalChannelParserTests.java | 40 +++++++++---------- .../config/xml/GatewayParserTests.java | 7 +++- .../core/AsyncMessagingTemplateTests.java | 6 ++- ...tOnceFileListFilterExternalStoreTests.java | 8 +++- ...rsistentAcceptOnceFileListFilterTests.java | 6 ++- .../lock/HazelcastLockRegistryTests.java | 25 ++++++++---- .../HttpRequestHandlingControllerTests.java | 7 +++- .../xml/KafkaOutboundAdapterParserTests.java | 9 +++-- .../util/AggregatorWithRedisLocksTests.java | 29 +++++++++----- .../redis/util/RedisLockRegistryTests.java | 20 +++++++--- .../outbound/SftpServerOutboundTests.java | 7 +++- .../zookeeper/lock/ZkLockRegistryTests.java | 16 ++++++-- 13 files changed, 132 insertions(+), 78 deletions(-) diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/FluxAggregatorMessageHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/FluxAggregatorMessageHandlerTests.java index c8ec685422..6ddbb41580 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/FluxAggregatorMessageHandlerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/FluxAggregatorMessageHandlerTests.java @@ -19,13 +19,13 @@ package org.springframework.integration.aggregator; import java.time.Duration; import java.util.List; import java.util.Objects; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.stream.Collectors; import java.util.stream.IntStream; import org.assertj.core.api.InstanceOfAssertFactories; import org.junit.jupiter.api.Test; -import org.junit.jupiter.api.condition.DisabledIfEnvironmentVariable; import reactor.core.publisher.Flux; import reactor.test.StepVerifier; @@ -156,8 +156,6 @@ class FluxAggregatorMessageHandlerTests { } @Test - @DisabledIfEnvironmentVariable(named = "bamboo_buildKey", matches = ".*?", - disabledReason = "Timing is too short for CI") void testWindowTimespan() { QueueChannel resultChannel = new QueueChannel(); FluxAggregatorMessageHandler fluxAggregatorMessageHandler = new FluxAggregatorMessageHandler(); @@ -165,18 +163,18 @@ class FluxAggregatorMessageHandlerTests { fluxAggregatorMessageHandler.setWindowTimespan(Duration.ofMillis(100)); fluxAggregatorMessageHandler.start(); - Executors.newSingleThreadExecutor() - .submit(() -> { - for (int i = 0; i < 10; i++) { - Message messageToAggregate = - MessageBuilder.withPayload(i) - .setCorrelationId("1") - .build(); - fluxAggregatorMessageHandler.handleMessage(messageToAggregate); - Thread.sleep(20); - } - return null; - }); + ExecutorService executorService = Executors.newSingleThreadExecutor(); + executorService.submit(() -> { + for (int i = 0; i < 10; i++) { + Message messageToAggregate = + MessageBuilder.withPayload(i) + .setCorrelationId("1") + .build(); + fluxAggregatorMessageHandler.handleMessage(messageToAggregate); + Thread.sleep(20); + } + return null; + }); Message result = resultChannel.receive(10_000); assertThat(result).isNotNull(); @@ -211,6 +209,8 @@ class FluxAggregatorMessageHandlerTests { .doesNotContain(0, 1); fluxAggregatorMessageHandler.stop(); + + executorService.shutdown(); } @Test diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/config/ThreadLocalChannelParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/config/ThreadLocalChannelParserTests.java index d392163d3f..b1d0ef5274 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/config/ThreadLocalChannelParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/config/ThreadLocalChannelParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2022 the original author or authors. + * Copyright 2002-2024 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. @@ -19,12 +19,11 @@ package org.springframework.integration.channel.config; import java.util.ArrayList; import java.util.List; import java.util.concurrent.CountDownLatch; -import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; -import org.junit.Test; -import org.junit.runner.RunWith; +import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; @@ -32,8 +31,7 @@ import org.springframework.integration.config.TestChannelInterceptor; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.support.GenericMessage; import org.springframework.test.annotation.DirtiesContext; -import org.springframework.test.context.ContextConfiguration; -import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; import static org.assertj.core.api.Assertions.assertThat; @@ -42,8 +40,7 @@ import static org.assertj.core.api.Assertions.assertThat; * @author Dave Syer * @author Artem Bilan */ -@RunWith(SpringJUnit4ClassRunner.class) -@ContextConfiguration +@SpringJUnitConfig @DirtiesContext public class ThreadLocalChannelParserTests { @@ -58,28 +55,29 @@ public class ThreadLocalChannelParserTests { @Test public void testSendInAnotherThread() throws Exception { - simpleChannel.send(new GenericMessage("test")); - Executor otherThreadExecutor = Executors.newSingleThreadExecutor(); + simpleChannel.send(new GenericMessage<>("test")); + ExecutorService otherThreadExecutor = Executors.newSingleThreadExecutor(); final CountDownLatch latch = new CountDownLatch(1); otherThreadExecutor.execute(() -> { - simpleChannel.send(new GenericMessage("crap")); + simpleChannel.send(new GenericMessage<>("crap")); latch.countDown(); }); assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); assertThat(simpleChannel.receive(10).getPayload()).isEqualTo("test"); // Message sent on another thread is not collected here - assertThat(simpleChannel.receive(10)).isEqualTo(null); + assertThat(simpleChannel.receive(1)).isEqualTo(null); + otherThreadExecutor.shutdown(); } @Test public void testReceiveInAnotherThread() throws Exception { - simpleChannel.send(new GenericMessage("test-1.1")); - simpleChannel.send(new GenericMessage("test-1.2")); - simpleChannel.send(new GenericMessage("test-1.3")); - channelWithInterceptor.send(new GenericMessage("test-2.1")); - channelWithInterceptor.send(new GenericMessage("test-2.2")); - Executor otherThreadExecutor = Executors.newSingleThreadExecutor(); - final List otherThreadResults = new ArrayList(); + simpleChannel.send(new GenericMessage<>("test-1.1")); + simpleChannel.send(new GenericMessage<>("test-1.2")); + simpleChannel.send(new GenericMessage<>("test-1.3")); + channelWithInterceptor.send(new GenericMessage<>("test-2.1")); + channelWithInterceptor.send(new GenericMessage<>("test-2.2")); + ExecutorService otherThreadExecutor = Executors.newSingleThreadExecutor(); + final List otherThreadResults = new ArrayList<>(); final CountDownLatch latch = new CountDownLatch(2); otherThreadExecutor.execute(() -> { otherThreadResults.add(simpleChannel.receive(0)); @@ -100,12 +98,14 @@ public class ThreadLocalChannelParserTests { assertThat(channelWithInterceptor.receive(0).getPayload()).isEqualTo("test-2.1"); assertThat(channelWithInterceptor.receive(0).getPayload()).isEqualTo("test-2.2"); assertThat(channelWithInterceptor.receive(0)).isNull(); + + otherThreadExecutor.shutdown(); } @Test public void testInterceptor() { int before = interceptor.getSendCount(); - channelWithInterceptor.send(new GenericMessage("test")); + channelWithInterceptor.send(new GenericMessage<>("test")); assertThat(interceptor.getSendCount()).isEqualTo(before + 1); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/GatewayParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/GatewayParserTests.java index 98a4f21f90..16f4d19ac2 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/GatewayParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/GatewayParserTests.java @@ -20,6 +20,7 @@ import java.time.Duration; import java.util.Map; import java.util.concurrent.Callable; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; @@ -395,7 +396,8 @@ public class GatewayParserTests { } private void startResponder(final PollableChannel requestChannel, final MessageChannel replyChannel) { - Executors.newSingleThreadExecutor().execute(() -> { + ExecutorService executorService = Executors.newSingleThreadExecutor(); + executorService.execute(() -> { Message request = requestChannel.receive(60000); assertThat(request).as("Request not received").isNotNull(); Message reply = MessageBuilder.fromMessage(request) @@ -405,7 +407,7 @@ public class GatewayParserTests { payload = CompletableFuture.completedFuture(reply); } else if (request.getPayload().equals("flowCompletable")) { - payload = CompletableFuture.completedFuture("SYNC_COMPLETABLE"); + payload = CompletableFuture.completedFuture("SYNC_COMPLETABLE"); } else if (request.getPayload().equals("flowCustomCompletable")) { MyCompletableFuture myCompletableFuture1 = new MyCompletableFuture(); @@ -427,6 +429,7 @@ public class GatewayParserTests { } replyChannel.send(reply); }); + executorService.shutdown(); } @SuppressWarnings("unused") diff --git a/spring-integration-core/src/test/java/org/springframework/integration/core/AsyncMessagingTemplateTests.java b/spring-integration-core/src/test/java/org/springframework/integration/core/AsyncMessagingTemplateTests.java index f23af0aeb4..d6ae4cd88d 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/core/AsyncMessagingTemplateTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/core/AsyncMessagingTemplateTests.java @@ -18,6 +18,7 @@ package org.springframework.integration.core; import java.util.concurrent.CancellationException; import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; @@ -432,8 +433,8 @@ public class AsyncMessagingTemplateTests { } private static void sendMessageAfterDelay(MessageChannel channel, GenericMessage message, int delay) { - Executors.newSingleThreadExecutor() - .execute(() -> { + ExecutorService executorService = Executors.newSingleThreadExecutor(); + executorService.execute(() -> { try { Thread.sleep(delay); } @@ -443,6 +444,7 @@ public class AsyncMessagingTemplateTests { } channel.send(message); }); + executorService.shutdown(); } private static class EchoHandler extends AbstractReplyProducingMessageHandler { diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/filters/PersistentAcceptOnceFileListFilterExternalStoreTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/filters/PersistentAcceptOnceFileListFilterExternalStoreTests.java index 02961189ea..af79905ef3 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/filters/PersistentAcceptOnceFileListFilterExternalStoreTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/filters/PersistentAcceptOnceFileListFilterExternalStoreTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2022 the original author or authors. + * Copyright 2014-2024 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. @@ -20,6 +20,7 @@ import java.io.File; import java.util.List; import java.util.Map; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; @@ -48,6 +49,7 @@ import static org.assertj.core.api.Assertions.assertThat; * @author Artem Bilan * @author Bojan Vukasovic * @author Artem Vozhdayenko + * * @since 4.0 */ public class PersistentAcceptOnceFileListFilterExternalStoreTests implements RedisContainerTest { @@ -139,7 +141,8 @@ public class PersistentAcceptOnceFileListFilterExternalStoreTests implements Red suspend.set(true); assertThat(file.setLastModified(file.lastModified() + 5000L)).isTrue(); - Future result = Executors.newSingleThreadExecutor() + ExecutorService executorService = Executors.newSingleThreadExecutor(); + Future result = executorService .submit(() -> filter.filterFiles(new File[] {file}).size()); assertThat(latch2.await(10, TimeUnit.SECONDS)).isTrue(); store.put("foo:" + file.getAbsolutePath(), "43"); @@ -149,6 +152,7 @@ public class PersistentAcceptOnceFileListFilterExternalStoreTests implements Red assertThat(file.delete()).isTrue(); filter.close(); + executorService.shutdown(); } } diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/filters/PersistentAcceptOnceFileListFilterTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/filters/PersistentAcceptOnceFileListFilterTests.java index b70ec87d00..5373503763 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/filters/PersistentAcceptOnceFileListFilterTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/filters/PersistentAcceptOnceFileListFilterTests.java @@ -22,6 +22,7 @@ import java.io.Flushable; import java.util.Arrays; import java.util.List; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; @@ -82,8 +83,8 @@ public class PersistentAcceptOnceFileListFilterTests extends AcceptOnceFileListF suspend.set(true); file.setLastModified(file.lastModified() + 5000L); - Future result = Executors.newSingleThreadExecutor() - .submit(() -> filter.filterFiles(new File[] {file}).size()); + ExecutorService executorService = Executors.newSingleThreadExecutor(); + Future result = executorService.submit(() -> filter.filterFiles(new File[] {file}).size()); assertThat(latch2.await(10, TimeUnit.SECONDS)).isTrue(); store.put("foo:" + file.getAbsolutePath(), "43"); latch1.countDown(); @@ -92,6 +93,7 @@ public class PersistentAcceptOnceFileListFilterTests extends AcceptOnceFileListF file.delete(); filter.close(); + executorService.shutdown(); } @Override diff --git a/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/lock/HazelcastLockRegistryTests.java b/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/lock/HazelcastLockRegistryTests.java index d254d8ec75..21346f39fd 100644 --- a/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/lock/HazelcastLockRegistryTests.java +++ b/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/lock/HazelcastLockRegistryTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2022 the original author or authors. + * Copyright 2017-2024 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,6 +17,7 @@ package org.springframework.integration.hazelcast.lock; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; @@ -28,8 +29,8 @@ import com.hazelcast.core.Hazelcast; import com.hazelcast.core.HazelcastInstance; import com.hazelcast.cp.lock.FencedLock; import com.hazelcast.instance.impl.HazelcastInstanceFactory; -import org.junit.AfterClass; -import org.junit.Test; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.Test; import static org.assertj.core.api.Assertions.assertThat; @@ -46,7 +47,7 @@ public class HazelcastLockRegistryTests { private static final HazelcastInstance instance = Hazelcast.newHazelcastInstance(CONFIG); - @AfterClass + @AfterAll public static void destroy() { HazelcastInstanceFactory.terminateAll(); } @@ -144,7 +145,8 @@ public class HazelcastLockRegistryTests { lock1.lockInterruptibly(); AtomicBoolean locked = new AtomicBoolean(); CountDownLatch latch = new CountDownLatch(1); - Future result = Executors.newSingleThreadExecutor().submit(() -> { + ExecutorService executorService = Executors.newSingleThreadExecutor(); + Future result = executorService.submit(() -> { Lock lock2 = registry.obtain("foo"); locked.set(lock2.tryLock(200, TimeUnit.MILLISECONDS)); latch.countDown(); @@ -162,6 +164,7 @@ public class HazelcastLockRegistryTests { Object ise = result.get(10, TimeUnit.SECONDS); assertThat(ise).isInstanceOf(IllegalMonitorStateException.class); assertThat(((Exception) ise).getMessage()).contains("Current thread is not owner of the lock!"); + executorService.shutdown(); } @Test @@ -173,7 +176,8 @@ public class HazelcastLockRegistryTests { CountDownLatch latch2 = new CountDownLatch(1); CountDownLatch latch3 = new CountDownLatch(1); lock1.lockInterruptibly(); - Executors.newSingleThreadExecutor().execute(() -> { + ExecutorService executorService = Executors.newSingleThreadExecutor(); + executorService.execute(() -> { Lock lock2 = registry.obtain("foo"); try { latch1.countDown(); @@ -195,6 +199,7 @@ public class HazelcastLockRegistryTests { latch2.countDown(); assertThat(latch3.await(10, TimeUnit.SECONDS)).isTrue(); assertThat(locked.get()).isTrue(); + executorService.shutdown(); } @Test @@ -207,7 +212,8 @@ public class HazelcastLockRegistryTests { CountDownLatch latch2 = new CountDownLatch(1); CountDownLatch latch3 = new CountDownLatch(1); lock1.lockInterruptibly(); - Executors.newSingleThreadExecutor().execute(() -> { + ExecutorService executorService = Executors.newSingleThreadExecutor(); + executorService.execute(() -> { Lock lock2 = registry2.obtain("foo"); try { latch1.countDown(); @@ -229,6 +235,7 @@ public class HazelcastLockRegistryTests { latch2.countDown(); assertThat(latch3.await(10, TimeUnit.SECONDS)).isTrue(); assertThat(locked.get()).isTrue(); + executorService.shutdown(); } @Test @@ -238,7 +245,8 @@ public class HazelcastLockRegistryTests { lock.lockInterruptibly(); final AtomicBoolean locked = new AtomicBoolean(); final CountDownLatch latch = new CountDownLatch(1); - Future result = Executors.newSingleThreadExecutor().submit(() -> { + ExecutorService executorService = Executors.newSingleThreadExecutor(); + Future result = executorService.submit(() -> { try { lock.unlock(); } @@ -254,6 +262,7 @@ public class HazelcastLockRegistryTests { Object imse = result.get(10, TimeUnit.SECONDS); assertThat(imse).isInstanceOf(IllegalMonitorStateException.class); assertThat(((Exception) imse).getMessage()).contains("Current thread is not owner of the lock!"); + executorService.shutdown(); } @Test diff --git a/spring-integration-http/src/test/java/org/springframework/integration/http/inbound/HttpRequestHandlingControllerTests.java b/spring-integration-http/src/test/java/org/springframework/integration/http/inbound/HttpRequestHandlingControllerTests.java index 5611a5b576..a938b885e2 100644 --- a/spring-integration-http/src/test/java/org/springframework/integration/http/inbound/HttpRequestHandlingControllerTests.java +++ b/spring-integration-http/src/test/java/org/springframework/integration/http/inbound/HttpRequestHandlingControllerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2022 the original author or authors. + * Copyright 2002-2024 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,6 +17,7 @@ package org.springframework.integration.http.inbound; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; @@ -357,7 +358,8 @@ public class HttpRequestHandlingControllerTests extends AbstractHttpInboundTests MockHttpServletResponse response = new MockHttpServletResponse(); final AtomicInteger active = new AtomicInteger(); final AtomicBoolean expected503 = new AtomicBoolean(); - Executors.newSingleThreadExecutor().execute(() -> { + ExecutorService executorService = Executors.newSingleThreadExecutor(); + executorService.execute(() -> { try { // wait for the active thread latch2.await(10, TimeUnit.SECONDS); @@ -387,6 +389,7 @@ public class HttpRequestHandlingControllerTests extends AbstractHttpInboundTests Object reply = modelAndView.getModel().get("reply"); assertThat(reply).isNotNull(); assertThat(reply).isEqualTo("HELLO"); + executorService.shutdown(); } @Test diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests.java index d5ba60362a..b60793bda8 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2022 the original author or authors. + * Copyright 2013-2024 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. @@ -19,6 +19,7 @@ package org.springframework.integration.kafka.config.xml; import java.time.Duration; import java.util.HashMap; import java.util.Map; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeoutException; @@ -125,8 +126,8 @@ class KafkaOutboundAdapterParserTests { handler.setTimeoutBuffer(200); handler.setTopicExpression(new LiteralExpression("foo")); - Executors.newSingleThreadExecutor() - .submit(() -> { + ExecutorService executorService = Executors.newSingleThreadExecutor(); + executorService.submit(() -> { RuntimeException exception = new RuntimeException("Async Producer Mock exception"); while (!mockProducer.errorNext(exception)) { Thread.sleep(100); @@ -146,6 +147,8 @@ class KafkaOutboundAdapterParserTests { .isThrownBy(() -> handler.handleMessage(new GenericMessage<>("foo"))) .withCauseInstanceOf(TimeoutException.class) .withStackTraceContaining("Timeout waiting for response from KafkaProducer"); + + executorService.shutdown(); } } diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/util/AggregatorWithRedisLocksTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/util/AggregatorWithRedisLocksTests.java index b49922895d..05e02dcd3a 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/util/AggregatorWithRedisLocksTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/util/AggregatorWithRedisLocksTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2022 the original author or authors. + * Copyright 2014-2024 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. @@ -20,6 +20,7 @@ import java.util.HashMap; import java.util.Map; import java.util.Set; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; @@ -88,8 +89,9 @@ class AggregatorWithRedisLocksTests implements RedisContainerTest { @Test void testLockSingleGroup() throws Exception { this.releaseStrategy.reset(1); - Executors.newSingleThreadExecutor().execute(asyncSend("foo", 1, 1)); - Executors.newSingleThreadExecutor().execute(asyncSend("bar", 2, 1)); + ExecutorService executorService = Executors.newCachedThreadPool(); + executorService.execute(asyncSend("foo", 1, 1)); + executorService.execute(asyncSend("bar", 2, 1)); assertThat(this.releaseStrategy.latch2.await(10, TimeUnit.SECONDS)).isTrue(); assertThat(this.template.keys("aggregatorWithRedisLocksTests:*")).hasSize(1); this.releaseStrategy.latch1.countDown(); @@ -98,17 +100,19 @@ class AggregatorWithRedisLocksTests implements RedisContainerTest { this.assertNoLocksAfterTest(); assertThat(this.exception) .as("Unexpected exception:" + (this.exception != null ? this.exception.toString() : "")).isNull(); + executorService.shutdown(); } @Test void testLockThreeGroups() throws Exception { this.releaseStrategy.reset(3); - Executors.newSingleThreadExecutor().execute(asyncSend("foo", 1, 1)); - Executors.newSingleThreadExecutor().execute(asyncSend("bar", 2, 1)); - Executors.newSingleThreadExecutor().execute(asyncSend("foo", 1, 2)); - Executors.newSingleThreadExecutor().execute(asyncSend("bar", 2, 2)); - Executors.newSingleThreadExecutor().execute(asyncSend("foo", 1, 3)); - Executors.newSingleThreadExecutor().execute(asyncSend("bar", 2, 3)); + ExecutorService executorService = Executors.newCachedThreadPool(); + executorService.execute(asyncSend("foo", 1, 1)); + executorService.execute(asyncSend("bar", 2, 1)); + executorService.execute(asyncSend("foo", 1, 2)); + executorService.execute(asyncSend("bar", 2, 2)); + executorService.execute(asyncSend("foo", 1, 3)); + executorService.execute(asyncSend("bar", 2, 3)); assertThat(this.releaseStrategy.latch2.await(10, TimeUnit.SECONDS)).isTrue(); assertThat(this.template.keys("aggregatorWithRedisLocksTests:*")).hasSize(3); this.releaseStrategy.latch1.countDown(); @@ -121,13 +125,15 @@ class AggregatorWithRedisLocksTests implements RedisContainerTest { this.assertNoLocksAfterTest(); assertThat(this.exception) .as("Unexpected exception:" + (this.exception != null ? this.exception.toString() : "")).isNull(); + executorService.shutdown(); } @RepeatedTest(10) void testDistributedAggregator() throws Exception { this.releaseStrategy.reset(1); - Executors.newSingleThreadExecutor().execute(asyncSend("foo", 1, 1)); - Executors.newSingleThreadExecutor().execute(() -> { + ExecutorService executorService = Executors.newCachedThreadPool(); + executorService.execute(asyncSend("foo", 1, 1)); + executorService.execute(() -> { try { in2.send(new GenericMessage<>("bar", stubHeaders(2, 2, 1))); } @@ -143,6 +149,7 @@ class AggregatorWithRedisLocksTests implements RedisContainerTest { this.assertNoLocksAfterTest(); assertThat(this.exception) .as("Unexpected exception:" + (this.exception != null ? this.exception.toString() : "")).isNull(); + executorService.shutdown(); } private void assertNoLocksAfterTest() throws Exception { diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/util/RedisLockRegistryTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/util/RedisLockRegistryTests.java index a0874de620..c654b6b85d 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/util/RedisLockRegistryTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/util/RedisLockRegistryTests.java @@ -228,7 +228,8 @@ class RedisLockRegistryTests implements RedisContainerTest { lock1.lockInterruptibly(); final AtomicBoolean locked = new AtomicBoolean(); final CountDownLatch latch = new CountDownLatch(1); - Future result = Executors.newSingleThreadExecutor().submit(() -> { + ExecutorService executorService = Executors.newSingleThreadExecutor(); + Future result = executorService.submit(() -> { Lock lock2 = registry.obtain("foo"); locked.set(lock2.tryLock(200, TimeUnit.MILLISECONDS)); latch.countDown(); @@ -249,6 +250,7 @@ class RedisLockRegistryTests implements RedisContainerTest { registry.expireUnusedOlderThan(-1000); assertThat(getRedisLockRegistryLocks(registry)).isEmpty(); registry.destroy(); + executorService.shutdown(); } @ParameterizedTest @@ -263,7 +265,8 @@ class RedisLockRegistryTests implements RedisContainerTest { CountDownLatch latch3 = new CountDownLatch(1); lock1.lockInterruptibly(); assertThat(getRedisLockRegistryLocks(registry)).hasSize(1); - Executors.newSingleThreadExecutor().execute(() -> { + ExecutorService executorService = Executors.newSingleThreadExecutor(); + executorService.execute(() -> { Lock lock2 = registry.obtain("foo"); try { latch1.countDown(); @@ -289,6 +292,7 @@ class RedisLockRegistryTests implements RedisContainerTest { registry.expireUnusedOlderThan(-1000); assertThat(getRedisLockRegistryLocks(registry)).isEmpty(); registry.destroy(); + executorService.shutdown(); } @ParameterizedTest @@ -305,7 +309,8 @@ class RedisLockRegistryTests implements RedisContainerTest { CountDownLatch latch3 = new CountDownLatch(1); lock1.lockInterruptibly(); assertThat(getRedisLockRegistryLocks(registry1)).hasSize(1); - Executors.newSingleThreadExecutor().execute(() -> { + ExecutorService executorService = Executors.newSingleThreadExecutor(); + executorService.execute(() -> { Lock lock2 = registry2.obtain("foo"); try { latch1.countDown(); @@ -340,6 +345,7 @@ class RedisLockRegistryTests implements RedisContainerTest { assertThat(getRedisLockRegistryLocks(registry2)).isEmpty(); registry1.destroy(); registry2.destroy(); + executorService.shutdown(); } @ParameterizedTest @@ -351,7 +357,8 @@ class RedisLockRegistryTests implements RedisContainerTest { lock.lockInterruptibly(); AtomicBoolean locked = new AtomicBoolean(); CountDownLatch latch = new CountDownLatch(1); - Future result = Executors.newSingleThreadExecutor().submit(() -> { + ExecutorService executorService = Executors.newSingleThreadExecutor(); + Future result = executorService.submit(() -> { try { lock.unlock(); } @@ -370,6 +377,7 @@ class RedisLockRegistryTests implements RedisContainerTest { registry.expireUnusedOlderThan(-1000); assertThat(getRedisLockRegistryLocks(registry)).isEmpty(); registry.destroy(); + executorService.shutdown(); } @ParameterizedTest @@ -481,7 +489,8 @@ class RedisLockRegistryTests implements RedisContainerTest { Long expire = getExpire(registry, "foo"); - Future result = Executors.newSingleThreadExecutor().submit(() -> { + ExecutorService executorService = Executors.newSingleThreadExecutor(); + Future result = executorService.submit(() -> { Lock lock2 = registry.obtain("foo"); assertThat(lock2.tryLock()).isFalse(); return null; @@ -490,6 +499,7 @@ class RedisLockRegistryTests implements RedisContainerTest { assertThat(getExpire(registry, "foo")).isEqualTo(expire); lock.unlock(); registry.destroy(); + executorService.shutdown(); } @ParameterizedTest diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests.java index 51924d064c..a1debf3dc9 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests.java +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests.java @@ -28,6 +28,7 @@ import java.io.UncheckedIOException; import java.util.ArrayList; import java.util.List; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.regex.Matcher; @@ -426,7 +427,8 @@ public class SftpServerOutboundTests extends SftpTestSupport { PipedOutputStream out2 = new PipedOutputStream(pipe2); final CountDownLatch latch1 = new CountDownLatch(1); final CountDownLatch latch2 = new CountDownLatch(1); - Executors.newSingleThreadExecutor().execute(() -> { + ExecutorService executorService = Executors.newSingleThreadExecutor(); + executorService.execute(() -> { try { session1.write(pipe1, "foo.txt"); } @@ -435,7 +437,7 @@ public class SftpServerOutboundTests extends SftpTestSupport { } latch1.countDown(); }); - Executors.newSingleThreadExecutor().execute(() -> { + executorService.execute(() -> { try { session2.write(pipe2, "bar.txt"); } @@ -465,6 +467,7 @@ public class SftpServerOutboundTests extends SftpTestSupport { session2.remove("bar.txt"); session1.close(); session2.close(); + executorService.shutdown(); } @Test diff --git a/spring-integration-zookeeper/src/test/java/org/springframework/integration/zookeeper/lock/ZkLockRegistryTests.java b/spring-integration-zookeeper/src/test/java/org/springframework/integration/zookeeper/lock/ZkLockRegistryTests.java index c4cbec63cf..e91590632c 100644 --- a/spring-integration-zookeeper/src/test/java/org/springframework/integration/zookeeper/lock/ZkLockRegistryTests.java +++ b/spring-integration-zookeeper/src/test/java/org/springframework/integration/zookeeper/lock/ZkLockRegistryTests.java @@ -146,7 +146,8 @@ public class ZkLockRegistryTests extends ZookeeperTestSupport { lock1.lockInterruptibly(); final AtomicBoolean locked = new AtomicBoolean(); final CountDownLatch latch = new CountDownLatch(1); - Future result = Executors.newSingleThreadExecutor().submit(() -> { + ExecutorService executorService = Executors.newSingleThreadExecutor(); + Future result = executorService.submit(() -> { Lock lock2 = registry.obtain("foo"); locked.set(lock2.tryLock(200, TimeUnit.MILLISECONDS)); latch.countDown(); @@ -165,6 +166,7 @@ public class ZkLockRegistryTests extends ZookeeperTestSupport { assertThat(ise).isInstanceOf(IllegalMonitorStateException.class); assertThat(((Exception) ise).getMessage()).contains("You do not own"); registry.destroy(); + executorService.shutdown(); } @Test @@ -176,7 +178,8 @@ public class ZkLockRegistryTests extends ZookeeperTestSupport { final CountDownLatch latch2 = new CountDownLatch(1); final CountDownLatch latch3 = new CountDownLatch(1); lock1.lockInterruptibly(); - Executors.newSingleThreadExecutor().execute(() -> { + ExecutorService executorService = Executors.newSingleThreadExecutor(); + executorService.execute(() -> { Lock lock2 = registry.obtain("foo"); try { latch1.countDown(); @@ -199,6 +202,7 @@ public class ZkLockRegistryTests extends ZookeeperTestSupport { assertThat(latch3.await(10, TimeUnit.SECONDS)).isTrue(); assertThat(locked.get()).isTrue(); registry.destroy(); + executorService.shutdown(); } @Test @@ -211,7 +215,8 @@ public class ZkLockRegistryTests extends ZookeeperTestSupport { final CountDownLatch latch2 = new CountDownLatch(1); final CountDownLatch latch3 = new CountDownLatch(1); lock1.lockInterruptibly(); - Executors.newSingleThreadExecutor().execute(() -> { + ExecutorService executorService = Executors.newSingleThreadExecutor(); + executorService.execute(() -> { Lock lock2 = registry2.obtain("foo"); try { latch1.countDown(); @@ -235,6 +240,7 @@ public class ZkLockRegistryTests extends ZookeeperTestSupport { assertThat(locked.get()).isTrue(); registry1.destroy(); registry2.destroy(); + executorService.shutdown(); } @Test @@ -244,7 +250,8 @@ public class ZkLockRegistryTests extends ZookeeperTestSupport { lock.lockInterruptibly(); final AtomicBoolean locked = new AtomicBoolean(); final CountDownLatch latch = new CountDownLatch(1); - Future result = Executors.newSingleThreadExecutor().submit(() -> { + ExecutorService executorService = Executors.newSingleThreadExecutor(); + Future result = executorService.submit(() -> { try { lock.unlock(); } @@ -261,6 +268,7 @@ public class ZkLockRegistryTests extends ZookeeperTestSupport { assertThat(imse).isInstanceOf(IllegalMonitorStateException.class); assertThat(((Exception) imse).getMessage()).contains("You do not own"); registry.destroy(); + executorService.shutdown(); } @Test