GH-3826: Fix SimplePool for resizing from MAX

Fixes https://github.com/spring-projects/spring-integration/issues/3826

By default, the `SimplePool` is used with an `Integer.MAX_VALUE` pool size.
There is a performance degradation in the `setPoolSize()` when we try to
decrease the pool size: the `while` for `permits.tryAcquire()` is too long
close to the current `Integer.MAX_VALUE` pool size

* Revise the logic in the `setPoolSize()` to use `Semaphore.reducePermits()`
instead of the loop.
* Change the calculation for a new pool size for the current pool state:
or it is a size of a new request, or iti s equal to the `inUse.size()`.
It will be reduced on subsequent `releaseItem()` calls
* Reduce the number of `available` according a new pool size based on the `inUse`.
So, if `inUse > newPoolSize`, the `available` is cleared.
Otherewise, it is reduced to the number which would give `newPoolSize` together
with the `inUse` size

**Cherry-pick to `5.5.x`**

spring-integration-core/src/main/java/org/springframework/integration/util/SimplePool.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2021 the original author or authors.
+ * Copyright 2002-2022 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.
@@ -48,7 +48,7 @@ public class SimplePool<T> implements Pool<T> {
 
 	protected final Log logger = LogFactory.getLog(getClass()); // NOSONAR final
 
-	private final Semaphore permits = new Semaphore(0);
+	private final PoolSemaphore permits = new PoolSemaphore(0);
 
 	private final AtomicInteger poolSize = new AtomicInteger();
 
@@ -56,11 +56,11 @@ public class SimplePool<T> implements Pool<T> {
 
 	private long waitTimeout = Long.MAX_VALUE;
 
-	private final BlockingQueue<T> available = new LinkedBlockingQueue<T>();
+	private final BlockingQueue<T> available = new LinkedBlockingQueue<>();
 
-	private final Set<T> allocated = Collections.synchronizedSet(new HashSet<T>());
+	private final Set<T> allocated = Collections.synchronizedSet(new HashSet<>());
 
-	private final Set<T> inUse = Collections.synchronizedSet(new HashSet<T>());
+	private final Set<T> inUse = Collections.synchronizedSet(new HashSet<>());
 
 	private final PoolItemCallback<T> callback;
 
@@ -105,21 +105,27 @@ public class SimplePool<T> implements Pool<T> {
 			this.permits.release(delta);
 		}
 		else {
-			while (delta < 0) {
+			this.permits.reducePermits(-delta);
-				if (!this.permits.tryAcquire()) {
+
-					break;
+			int inUseSize = this.inUse.size();
-				}
+			int newPoolSize = Math.max(poolSize, inUseSize);
+			this.poolSize.set(newPoolSize);
+
+			for (int i = this.available.size(); i > newPoolSize - inUseSize; i--) {
 				T item = this.available.poll();
 				if (item != null) {
 					doRemoveItem(item);
 				}
-				this.poolSize.decrementAndGet();
+				else {
-				delta++;
+					break;
+				}
+			}
+
+			int inUseDelta = poolSize - inUseSize;
+			if (inUseDelta < 0 && this.logger.isDebugEnabled()) {
+				this.logger.debug(String.format("Pool is overcommitted by %d; items will be removed when returned",
+						-inUseDelta));
 			}
-		}
-		if (delta < 0 && this.logger.isDebugEnabled()) {
-			this.logger.debug(String.format("Pool is overcommitted by %d; items will be removed when returned",
-					-delta));
 		}
 	}
 
@@ -266,6 +272,19 @@ public class SimplePool<T> implements Pool<T> {
 		removeAllIdleItems();
 	}
 
+	private static class PoolSemaphore extends Semaphore {
+
+		PoolSemaphore(int permits) {
+			super(permits);
+		}
+
+		@Override
+		public void reducePermits(int reduction) { // NOSONAR increases visibility
+			super.reducePermits(reduction);
+		}
+
+	}
+
 	/**
 	 * User of the pool provide an implementation of this interface; called during
 	 * various pool operations.

spring-integration-core/src/test/java/org/springframework/integration/util/SimplePoolTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2020 the original author or authors.
+ * Copyright 2002-2022 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,7 +20,6 @@ import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
 import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
 import static org.assertj.core.api.Assertions.assertThatIllegalStateException;
-import static org.assertj.core.api.Assertions.fail;
 
 import java.util.ArrayList;
 import java.util.HashSet;
@@ -36,6 +35,8 @@ import org.springframework.integration.test.util.TestUtils;
 /**
  * @author Gary Russell
  * @author Sergey Bogatyrev
+ * @author Artem Bilan
+ *
  * @since 2.2
  *
  */
@@ -43,7 +44,7 @@ public class SimplePoolTests {
 
 	@Test
 	public void testReuseAndStale() {
-		final Set<String> strings = new HashSet<String>();
+		final Set<String> strings = new HashSet<>();
 		final AtomicBoolean stale = new AtomicBoolean();
 		SimplePool<String> pool = stringPool(2, strings, stale);
 		String s1 = pool.getItem();
@@ -62,7 +63,7 @@ public class SimplePoolTests {
 
 	@Test
 	public void testOverCommitAndResize() {
-		final Set<String> strings = new HashSet<String>();
+		final Set<String> strings = new HashSet<>();
 		final AtomicBoolean stale = new AtomicBoolean();
 		SimplePool<String> pool = stringPool(2, strings, stale);
 		String s1 = pool.getItem();
@@ -83,13 +84,9 @@ public class SimplePoolTests {
 		assertThat(pool.getIdleCount()).isEqualTo(0);
 		assertThat(pool.getActiveCount()).isEqualTo(2);
 		assertThat(pool.getAllocatedCount()).isEqualTo(2);
-		try {
-			pool.getItem();
-			fail("Expected exception");
-		}
-		catch (PoolItemNotAvailableException e) {
 
-		}
+		assertThatExceptionOfType(PoolItemNotAvailableException.class)
+				.isThrownBy(pool::getItem);
 
 		// resize up
 		pool.setPoolSize(4);
@@ -131,7 +128,7 @@ public class SimplePoolTests {
 
 	@Test
 	public void testForeignObject() {
-		final Set<String> strings = new HashSet<String>();
+		final Set<String> strings = new HashSet<>();
 		final AtomicBoolean stale = new AtomicBoolean();
 		SimplePool<String> pool = stringPool(2, strings, stale);
 		pool.getItem();
@@ -140,7 +137,7 @@ public class SimplePoolTests {
 
 	@Test
 	public void testDoubleReturn() {
-		final Set<String> strings = new HashSet<String>();
+		final Set<String> strings = new HashSet<>();
 		final AtomicBoolean stale = new AtomicBoolean();
 		SimplePool<String> pool = stringPool(2, strings, stale);
 		Semaphore permits = TestUtils.getPropertyValue(pool, "permits", Semaphore.class);
@@ -168,7 +165,27 @@ public class SimplePoolTests {
 		assertThat(allocatedItems).hasSize(5);
 
 		// no more items can be allocated (indirect check of permits)
-		assertThatExceptionOfType(PoolItemNotAvailableException.class).isThrownBy(() -> pool.getItem());
+		assertThatExceptionOfType(PoolItemNotAvailableException.class)
+				.isThrownBy(pool::getItem);
+	}
+
+	@Test
+	public void testMaxValueSizeUpdateIfNotAllocated() {
+		SimplePool<String> pool = stringPool(0, new HashSet<>(), new AtomicBoolean());
+		pool.setWaitTimeout(0);
+		pool.setPoolSize(5);
+		assertThat(pool.getPoolSize()).isEqualTo(5);
+
+		// allocating all available items to check permits
+		Set<String> allocatedItems = new HashSet<>();
+		for (int i = 0; i < 5; i++) {
+			allocatedItems.add(pool.getItem());
+		}
+		assertThat(allocatedItems).hasSize(5);
+
+		// no more items can be allocated (indirect check of permits)
+		assertThatExceptionOfType(PoolItemNotAvailableException.class)
+				.isThrownBy(pool::getItem);
 	}
 
 	@Test
@@ -207,7 +224,54 @@ public class SimplePoolTests {
 		assertThat(pool.getActiveCount()).isEqualTo(5);
 
 		// no more items can be allocated (indirect check of permits)
-		assertThatExceptionOfType(PoolItemNotAvailableException.class).isThrownBy(() -> pool.getItem());
+		assertThatExceptionOfType(PoolItemNotAvailableException.class)
+				.isThrownBy(pool::getItem);
+	}
+
+	@Test
+	public void testMaxValueSizeUpdateIfPartiallyAllocated() {
+		SimplePool<String> pool = stringPool(0, new HashSet<>(), new AtomicBoolean());
+		pool.setWaitTimeout(0);
+
+		List<String> allocated = new ArrayList<>();
+		for (int i = 0; i < 10; i++) {
+			allocated.add(pool.getItem());
+		}
+
+		// release only 2 items
+		for (int i = 0; i < 2; i++) {
+			pool.releaseItem(allocated.get(i));
+		}
+
+		// release only 2 items
+		for (int i = 0; i < 2; i++) {
+			pool.releaseItem(allocated.get(i));
+		}
+
+		// trying to reduce pool size
+		pool.setPoolSize(5);
+
+		// at this moment the actual pool size can be reduced only partially, because
+		// only 2 items have been released, so 8 items are in use
+		assertThat(pool.getPoolSize()).isEqualTo(8);
+		assertThat(pool.getAllocatedCount()).isEqualTo(8);
+		assertThat(pool.getIdleCount()).isEqualTo(0);
+		assertThat(pool.getActiveCount()).isEqualTo(8);
+
+		// releasing 3 items
+		for (int i = 2; i < 5; i++) {
+			pool.releaseItem(allocated.get(i));
+		}
+
+		// now pool size should be reduced
+		assertThat(pool.getPoolSize()).isEqualTo(5);
+		assertThat(pool.getAllocatedCount()).isEqualTo(5);
+		assertThat(pool.getIdleCount()).isEqualTo(0);
+		assertThat(pool.getActiveCount()).isEqualTo(5);
+
+		// no more items can be allocated (indirect check of permits)
+		assertThatExceptionOfType(PoolItemNotAvailableException.class)
+				.isThrownBy(pool::getItem);
 	}
 
 	@Test
@@ -240,7 +304,8 @@ public class SimplePoolTests {
 		assertThat(pool.getActiveCount()).isEqualTo(5);
 
 		// no more items can be allocated (indirect check of permits)
-		assertThatExceptionOfType(PoolItemNotAvailableException.class).isThrownBy(() -> pool.getItem());
+		assertThatExceptionOfType(PoolItemNotAvailableException.class)
+				.isThrownBy(pool::getItem);
 
 		// releasing remaining items
 		for (int i = 5; i < 10; i++) {
@@ -266,10 +331,9 @@ public class SimplePoolTests {
 		assertThatIllegalStateException().isThrownBy(pool::getItem);
 	}
 
-	private SimplePool<String> stringPool(int size, final Set<String> strings,
+	private SimplePool<String> stringPool(int size, Set<String> strings, AtomicBoolean stale) {
-			final AtomicBoolean stale) {
+		return new SimplePool<String>(size, new SimplePool.PoolItemCallback<String>() {
 
-		SimplePool<String> pool = new SimplePool<String>(size, new SimplePool.PoolItemCallback<String>() {
 			private int i;
 
 			@Override
@@ -293,7 +357,6 @@ public class SimplePoolTests {
 			}
 
 		});
-		return pool;
 	}
 
 }