From 8df8028537f06acca3b833f8442f8560e1ab5c8f Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 8 Jul 2020 14:19:57 -0400 Subject: [PATCH] Fix CachedSessionFactory Race Close the pool so that any sessions returned after the factory is `destroy()`ed are closed. * Call `removeAllIdleItems()` in `close()`. * Close sessions in `SftpStreamingMessageSourceTests`. **cherry-pick to all supported branches** # Conflicts: # spring-integration-core/src/test/java/org/springframework/integration/util/SimplePoolTests.java --- .../integration/util/Pool.java | 7 ++++ .../integration/util/SimplePool.java | 11 +++++- .../integration/util/SimplePoolTests.java | 37 +++++++++++++++++-- .../remote/session/CachingSessionFactory.java | 2 +- .../CachingClientConnectionFactory.java | 8 +++- .../SftpStreamingMessageSourceTests.java | 4 ++ 6 files changed, 63 insertions(+), 6 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/util/Pool.java b/spring-integration-core/src/main/java/org/springframework/integration/util/Pool.java index 8a201d0214..351ae22f67 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/util/Pool.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/util/Pool.java @@ -72,4 +72,11 @@ public interface Pool { */ int getAllocatedCount(); + /** + * Close the pool; returned items will be destroyed. + * @since 4.3.23 + */ + default void close() { + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/util/SimplePool.java b/spring-integration-core/src/main/java/org/springframework/integration/util/SimplePool.java index b8010d80a0..0a1bb5c32c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/util/SimplePool.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/util/SimplePool.java @@ -59,6 +59,8 @@ public class SimplePool implements Pool { private final PoolItemCallback callback; + private volatile boolean closed; + /** * Creates a SimplePool with a specific limit. * @param poolSize The maximum number of items the pool supports. @@ -158,6 +160,7 @@ public class SimplePool implements Pool { */ @Override public T getItem() { + Assert.state(!this.closed, "Pool has been closed"); boolean permitted = false; try { try { @@ -215,7 +218,7 @@ public class SimplePool implements Pool { Assert.isTrue(this.allocated.contains(item), "You can only release items that were obtained from the pool"); if (this.inUse.contains(item)) { - if (this.poolSize.get() > this.targetPoolSize.get()) { + if (this.poolSize.get() > this.targetPoolSize.get() || this.closed) { this.poolSize.decrementAndGet(); if (item != null) { doRemoveItem(item); @@ -256,6 +259,12 @@ public class SimplePool implements Pool { this.callback.removedFromPool(item); } + @Override + public synchronized void close() { + this.closed = true; + removeAllIdleItems(); + } + /** * User of the pool provide an implementation of this interface; called during * various pool operations. diff --git a/spring-integration-core/src/test/java/org/springframework/integration/util/SimplePoolTests.java b/spring-integration-core/src/test/java/org/springframework/integration/util/SimplePoolTests.java index 832fdaff79..c676fcd789 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/util/SimplePoolTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/util/SimplePoolTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2020 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. @@ -16,10 +16,12 @@ package org.springframework.integration.util; +import static org.hamcrest.Matchers.instanceOf; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotSame; import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThat; import static org.junit.Assert.fail; import java.util.HashSet; @@ -27,6 +29,7 @@ import java.util.Set; import java.util.concurrent.Semaphore; import java.util.concurrent.atomic.AtomicBoolean; + import org.junit.Test; import org.springframework.integration.test.util.TestUtils; @@ -124,13 +127,18 @@ public class SimplePoolTests { assertEquals(2, pool.getAllocatedCount()); } - @Test(expected = IllegalArgumentException.class) + @Test public void testForeignObject() { final Set strings = new HashSet(); final AtomicBoolean stale = new AtomicBoolean(); SimplePool pool = stringPool(2, strings, stale); pool.getItem(); - pool.releaseItem("Hello, world!"); + try { + pool.releaseItem("Hello, world!"); + } + catch (Exception e) { + assertThat(e, instanceOf(IllegalArgumentException.class)); + } } @Test @@ -149,16 +157,37 @@ public class SimplePoolTests { } + @Test + public void testClose() { + SimplePool pool = stringPool(10, new HashSet<>(), new AtomicBoolean()); + String item1 = pool.getItem(); + String item2 = pool.getItem(); + pool.releaseItem(item2); + assertEquals(2, pool.getAllocatedCount()); + pool.close(); + pool.releaseItem(item1); + assertEquals(0, pool.getAllocatedCount()); + try { + pool.getItem(); + } + catch (Exception e) { + assertThat(e, instanceOf(IllegalStateException.class)); + } + } + private SimplePool stringPool(int size, final Set strings, final AtomicBoolean stale) { + SimplePool pool = new SimplePool(size, new SimplePool.PoolItemCallback() { private int i; + @Override public String createForPool() { String string = "String" + i++; strings.add(string); return string; } + @Override public boolean isStale(String item) { if (stale.get()) { @@ -166,10 +195,12 @@ public class SimplePoolTests { } return stale.get(); } + @Override public void removedFromPool(String item) { strings.remove(item); } + }); return pool; } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/CachingSessionFactory.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/CachingSessionFactory.java index db37ae1aeb..64d0375fcd 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/CachingSessionFactory.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/CachingSessionFactory.java @@ -140,7 +140,7 @@ public class CachingSessionFactory implements SessionFactory, DisposableBe */ @Override public void destroy() { - this.pool.removeAllIdleItems(); + this.pool.close(); } /** diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java index f28729b2fe..751ee78c5b 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java @@ -21,6 +21,7 @@ import java.util.Map; import java.util.concurrent.Executor; import java.util.concurrent.atomic.AtomicBoolean; +import org.springframework.beans.factory.DisposableBean; import org.springframework.core.serializer.Deserializer; import org.springframework.core.serializer.Serializer; import org.springframework.integration.ip.IpHeaders; @@ -41,7 +42,7 @@ import org.springframework.messaging.support.ErrorMessage; * @since 2.2 * */ -public class CachingClientConnectionFactory extends AbstractClientConnectionFactory { +public class CachingClientConnectionFactory extends AbstractClientConnectionFactory implements DisposableBean { private final AbstractClientConnectionFactory targetConnectionFactory; @@ -389,6 +390,11 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact this.pool.removeAllIdleItems(); } + @Override + public void destroy() throws Exception { + this.pool.close(); + } + private final class CachedConnection extends TcpConnectionInterceptorSupport { private final AtomicBoolean released = new AtomicBoolean(); diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java index b1221471dd..f4415e4253 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java @@ -36,6 +36,7 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.ApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import org.springframework.integration.StaticMessageHeaderAccessor; import org.springframework.integration.annotation.InboundChannelAdapter; import org.springframework.integration.annotation.Transformer; import org.springframework.integration.channel.QueueChannel; @@ -138,6 +139,7 @@ public class SftpStreamingMessageSourceTests extends SftpTestSupport { anyOf(equalTo(" sftpSource1.txt"), equalTo("sftpSource2.txt"))); received.getPayload().close(); + StaticMessageHeaderAccessor.getCloseableResource(received).close(); } @Test @@ -152,6 +154,7 @@ public class SftpStreamingMessageSourceTests extends SftpTestSupport { anyOf(equalTo(" sftpSource1.txt"), equalTo("sftpSource2.txt"))); received.getPayload().close(); + StaticMessageHeaderAccessor.getCloseableResource(received).close(); } @Test @@ -166,6 +169,7 @@ public class SftpStreamingMessageSourceTests extends SftpTestSupport { anyOf(equalTo(" sftpSource1.txt"), equalTo("sftpSource2.txt"))); received.getPayload().close(); + StaticMessageHeaderAccessor.getCloseableResource(received).close(); } private SftpStreamingMessageSource buildSource() {