Repository: quarkusio/quarkus
Issue: ConcurrentModificationException  thrown when closing a connection on a reactive pool

**Describe the bug**
While operating database, an exceptions is thrown randomly (but very often), here's the stacktrace:
```
	Caused by: java.util.ConcurrentModificationException
		at java.base/java.util.ArrayList$Itr.checkForComodification(ArrayList.java:1013)
		at java.base/java.util.ArrayList$Itr.next(ArrayList.java:967)
		at io.quarkus.reactive.datasource.runtime.ThreadLocalPool.scanForAbandonedConnections(ThreadLocalPool.java:75)
		at io.quarkus.reactive.datasource.runtime.ThreadLocalPool.pool(ThreadLocalPool.java:68)
		at io.quarkus.reactive.datasource.runtime.ThreadLocalPool.preparedQuery(ThreadLocalPool.java:108)
		at io.vertx.mutiny.sqlclient.Pool.preparedQuery(Pool.java:149)
		at io.vertx.mutiny.pgclient.PgPool_2d3061477c44a6b700dc3f9c4fe453a846636193_Synthetic_ClientProxy.preparedQuery(PgPool_2d3061477c44a6b700dc3f9c4fe453a846636193_Synthetic_ClientProxy.zig:275)
		at com.buguroo.environment.traveler.service.dao.UserCountryDao.forCompanyAndUser(UserCountryDao.java:44)
		at com.buguroo.environment.traveler.service.dao.UserCountryDao_Subclass.forCompanyAndUser$$superaccessor3(UserCountryDao_Subclass.zig:589)
		at com.buguroo.environment.traveler.service.dao.UserCountryDao_Subclass$$function$$3.apply(UserCountryDao_Subclass$$function$$3.zig:41)
		at io.quarkus.arc.impl.AroundInvokeInvocationContext.proceed(AroundInvokeInvocationContext.java:54)
		at io.quarkus.narayana.jta.runtime.interceptor.TransactionalInterceptorBase.invokeInOurTx(TransactionalInterceptorBase.java:127)
		at io.quarkus.narayana.jta.runtime.interceptor.TransactionalInterceptorBase.invokeInOurTx(TransactionalInterceptorBase.java:100)
		at io.quarkus.narayana.jta.runtime.interceptor.TransactionalInterceptorRequired.doIntercept(TransactionalInterceptorRequired.java:32)
		at io.quarkus.narayana.jta.runtime.interceptor.TransactionalInterceptorBase.intercept(TransactionalInterceptorBase.java:53)
		at io.quarkus.narayana.jta.runtime.interceptor.TransactionalInterceptorRequired.intercept(TransactionalInterceptorRequired.java:26)
		at io.quarkus.narayana.jta.runtime.interceptor.TransactionalInterceptorRequired_Bean.intercept(TransactionalInterceptorRequired_Bean.zig:340)
		at io.quarkus.arc.impl.InterceptorInvocation.invoke(InterceptorInvocation.java:41)
		at io.quarkus.arc.impl.AroundInvokeInvocationContext.perform(AroundInvokeInvocationContext.java:41)
		at io.quarkus.arc.impl.InvocationContexts.performAroundInvoke(InvocationContexts.java:32)
		at com.buguroo.environment.traveler.service.dao.UserCountryDao_Subclass.forCompanyAndUser(UserCountryDao_Subclass.zig:544)
		at com.buguroo.environment.traveler.service.dao.UserCountryDao_ClientProxy.forCompanyAndUser(UserCountryDao_ClientProxy.zig:191)
		at com.buguroo.environment.traveler.service.FrequentTravelerService.getOpinion(FrequentTravelerService.java:39)
		at com.buguroo.environment.traveler.service.FrequentTravelerService_ClientProxy.getOpinion(FrequentTravelerService_ClientProxy.zig:157)
		at com.buguroo.environment.traveler.FrequentTravelerListener.process(FrequentTravelerListener.java:57)
		at com.buguroo.environment.traveler.FrequentTravelerListener_Subclass.process$$superaccessor1(FrequentTravelerListener_Subclass.zig:400)
		at com.buguroo.environment.traveler.FrequentTravelerListener_Subclass$$function$$1.apply(FrequentTravelerListener_Subclass$$function$$1.zig:33)
		at io.quarkus.arc.impl.AroundInvokeInvocationContext.proceed(AroundInvokeInvocationContext.java:54)
		at io.smallrye.faulttolerance.FaultToleranceInterceptor.lambda$null$1(FaultToleranceInterceptor.java:178)
		... 6 more
```

From what it seems, the code was introduced in https://github.com/quarkusio/quarkus/pull/14134 . In fact the behaviour disappears if I turn back to 1.10.5.
The PR contains a test, but the test does not cover the scenario because it fails only when the "dead" connection is not the last (because it calls close which ends up modifying `allConnections`).

**Expected behavior**
Not to throw an exception ;).

However, I'm wondering if I'm doing something wrong in my code because I wouldn't expect pools to be closed this often.

**Actual behavior**
The added exception is thrown.


**To Reproduce**

Link to a small reproducer (preferably a Maven project if the issue is not Gradle-specific).

Or attach an archive containing the reproducer to the issue.

Steps to reproduce the behavior:
1. 
2. 
3. 

**Configuration**
Configuring reactive database connection pooling.

**Screenshots**
N/A

**Environment (please complete the following information):**
 - Output of `uname -a` or `ver`: Linux <redacted> 5.4.0-54-generic #60-Ubuntu SMP Fri Nov 6 10:37:59 UTC 2020 x86_64 x86_64 x86_64 GNU/Linux
 - Output of `java -version`: 
 ```
openjdk version "14.0.2" 2020-07-14
OpenJDK Runtime Environment (build 14.0.2+12-46)
OpenJDK 64-Bit Server VM (build 14.0.2+12-46, mixed mode, sharing)
```
 - GraalVM version (if different from Java): 
 - Quarkus version or git rev: 1.11.0
 - Build tool (ie. output of `mvnw --version` or `gradlew --version`): 
```
Apache Maven 3.6.3 (cecedd343002696d0abb50b32b541b8a6ba2883f)
Maven home: /home/jose/.m2/wrapper/dists/apache-maven-3.6.3-bin/1iopthnavndlasol9gbrbg6bf2/apache-maven-3.6.3
Java version: 14.0.2, vendor: Oracle Corporation, runtime: /home/jose/.sdkman/candidates/java/14.0.2-open
Default locale: en_US, platform encoding: UTF-8
OS name: "linux", version: "5.4.0-54-generic", arch: "amd64", family: "unix"
```

**Additional context**
N/A


Changed files:
extensions/reactive-datasource/runtime/src/main/java/io/quarkus/reactive/datasource/runtime/ThreadLocalPool.java
extensions/reactive-datasource/deployment/src/test/java/io/quarkus/reactive/datasource/runtime/ConnectionPoolsClosedTest.java

diff --git a/extensions/reactive-datasource/deployment/src/test/java/io/quarkus/reactive/datasource/runtime/ConnectionPoolsClosedTest.java b/extensions/reactive-datasource/deployment/src/test/java/io/quarkus/reactive/datasource/runtime/ConnectionPoolsClosedTest.java
index ef3892b0e94..7dc8bec3e87 100644
--- a/extensions/reactive-datasource/deployment/src/test/java/io/quarkus/reactive/datasource/runtime/ConnectionPoolsClosedTest.java
+++ b/extensions/reactive-datasource/deployment/src/test/java/io/quarkus/reactive/datasource/runtime/ConnectionPoolsClosedTest.java
@@ -40,6 +40,45 @@ public void connectionsFromOtherThreadsGetClosed() throws ExecutionException, In
         Assert.assertTrue(globalPool.trackedSize() == 0);
     }
 
+    /**
+     * This test makes sure that when the closed connection is not the last one in the list
+     * of closed connections, no exception is thrown.
+     */
+    @Test
+    public void connectionsThatAreNotTheLastGetClosedSuccessfully()
+            throws ExecutionException, InterruptedException {
+        ExecutorService e = Executors.newFixedThreadPool(3);
+        try {
+            TestableThreadLocalPool globalPool = new TestableThreadLocalPool();
+            Assert.assertTrue(globalPool.trackedSize() == 0);
+            final TestPoolInterface p1 = grabPoolFromLongLivingThread(globalPool, e);
+            Assert.assertFalse(p1.isClosed());
+            Assert.assertTrue(globalPool.trackedSize() == 1);
+            final TestPoolInterface p2 = grabPoolFromLongLivingThread(globalPool, e);
+            Assert.assertFalse(p1.isClosed());
+            Assert.assertFalse(p2.isClosed());
+            Assert.assertFalse(p1.isClosed());
+            Assert.assertFalse(p2.isClosed());
+            Assert.assertTrue(globalPool.trackedSize() == 2);
+            e.shutdown();
+            while (!e.isTerminated()) {
+                Thread.sleep(1);
+            }
+            final TestPoolInterface p3 = grabPoolFromOtherThread(globalPool);
+            Assert.assertTrue(p1.isClosed());
+            Assert.assertTrue(p2.isClosed());
+            Assert.assertFalse(p3.isClosed());
+            Assert.assertTrue(globalPool.trackedSize() == 1);
+            globalPool.close();
+            Assert.assertTrue(p1.isClosed());
+            Assert.assertTrue(p2.isClosed());
+            Assert.assertTrue(p3.isClosed());
+            Assert.assertTrue(globalPool.trackedSize() == 0);
+        } finally {
+            e.shutdown();
+        }
+    }
+
     /**
      * Here we check that when explicit close of a thread-local
      * specific pool is closed, we also de-reference it.
@@ -84,4 +123,15 @@ private TestPoolInterface grabPoolFromOtherThread(TestableThreadLocalPool global
         return poolInstance;
     }
 
+    private TestPoolInterface grabPoolFromLongLivingThread(TestableThreadLocalPool globalPool,
+            ExecutorService e)
+            throws InterruptedException, ExecutionException {
+        AtomicReference<Thread> thread = new AtomicReference<>();
+        final Future<TestPoolInterface> creation = e.submit(() -> {
+            thread.set(Thread.currentThread());
+            return globalPool.pool();
+        });
+        return creation.get();
+    }
+
 }
diff --git a/extensions/reactive-datasource/runtime/src/main/java/io/quarkus/reactive/datasource/runtime/ThreadLocalPool.java b/extensions/reactive-datasource/runtime/src/main/java/io/quarkus/reactive/datasource/runtime/ThreadLocalPool.java
index 5ad8da4b20c..4474217ef2b 100644
--- a/extensions/reactive-datasource/runtime/src/main/java/io/quarkus/reactive/datasource/runtime/ThreadLocalPool.java
+++ b/extensions/reactive-datasource/runtime/src/main/java/io/quarkus/reactive/datasource/runtime/ThreadLocalPool.java
@@ -72,13 +72,19 @@ PoolType pool() {
     }
 
     private final void scanForAbandonedConnections() {
+        ArrayList<PoolAndThread> garbage = new ArrayList<>();
         for (PoolAndThread pair : allConnections) {
             if (pair.isDead()) {
-                //This might potentially close the connection a second time,
-                //so we need to ensure implementations allow it.
-                pair.close();
+                garbage.add(pair);
             }
         }
+        //This needs a second loop, as the close() operation
+        //will otherwise trigger a concurrent modification on the iterator.
+        for (PoolAndThread dead : garbage) {
+            //This might potentially close the connection a second time,
+            //so we need to ensure implementations allow it.
+            dead.close();
+        }
     }
 
     private void checkPoolIsOpen() {