From c95e6dcc3b4c03570cb7d3065c8cecadd879a48a 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 # Conflicts: # spring-integration-core/src/test/java/org/springframework/integration/util/SimplePoolTests.java # spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java # spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java * Remove `default close()` from `Pool` interface * Cast to `SimplePool` in the `CSF` to call the new `close()` --- .../integration/util/SimplePool.java | 10 +++++- .../integration/util/SimplePoolTests.java | 34 +++++++++++++++++-- .../remote/session/CachingSessionFactory.java | 2 +- .../CachingClientConnectionFactory.java | 12 ++++++- 4 files changed, 52 insertions(+), 6 deletions(-) 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 11f353f938..644e2ed928 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 @@ -60,6 +60,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. @@ -154,6 +156,7 @@ public class SimplePool implements Pool { * @throws MessagingException if no items become available in time. */ public T getItem() { + Assert.state(!this.closed, "Pool has been closed"); boolean permitted = false; try { try { @@ -210,7 +213,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); @@ -250,6 +253,11 @@ public class SimplePool implements Pool { this.callback.removedFromPool(item); } + 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 9323847eb1..e751dba65f 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-2016 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; @@ -125,13 +128,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 @@ -150,8 +158,27 @@ 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; public String createForPool() { @@ -168,6 +195,7 @@ public class SimplePoolTests { 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 ddf0ae177c..7047fff422 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 @@ -128,7 +128,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 bc6c7f3e49..016a74763c 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; @@ -40,7 +41,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; @@ -403,6 +404,15 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact this.targetConnectionFactory.stop(callback); } + @Override + public void destroy() throws Exception { + if (this.pool instanceof SimplePool) { + ((SimplePool) this.pool).close(); + } + } + + + private final class CachedConnection extends TcpConnectionInterceptorSupport { private final AtomicBoolean released = new AtomicBoolean();