Fix lease timeout race to prevent pool entry leak

[arturobernalg]

No description provided.

[ok2c]

@arturobernalg You see how much smaller and cleaner the fix is when applied at the right place. Let me poke at your changes a bit more over the weekend but fundamentally looks OK.

[arturobernalg]

You see how much smaller and cleaner the fix is when applied at the right place. Let me poke at your changes a bit more over the weekend but fundamentally looks OK.

@arturobernalg You see how much smaller and cleaner the fix is when applied at the right place. Let me poke at your changes a bit more over the weekend but fundamentally looks OK.

@ok2c Indeed. make more sense

[ok2c]

@arturobernalg Can we also try to get this test case to work?

Interestingly enough, OFFLOCK is the only pool implementation that appears to be able to handle the test correctly. Kudos to you, But it can also be my test case is flawed, so let's be diligent.

    @ParameterizedTest(name = "{0}")
    @MethodSource("pools")
    @org.junit.jupiter.api.Timeout(60)
    void testInterruptDoesNotLeakLeasedEntries(final PoolCase poolCase) throws Exception {
        final ManagedConnPool<String, DummyConn> pool = poolCase.supplier.get();

        final String route = "route-1";

        final int concurrentThreads = 10;

        final ExecutorService executorService = Executors.newFixedThreadPool(concurrentThreads);
        final AtomicReference<Exception> unexpectedException = new AtomicReference<>();
        try {
            final Queue<Future<?>> taskQueue = new ConcurrentLinkedQueue<>();
            for (int i = 0; i < concurrentThreads * 100; i++) {
                taskQueue.add(executorService.submit(() -> {
                    final Future<PoolEntry<String, DummyConn>> f = pool.lease(route, null, TIMEOUT, null);
                    try {
                        final PoolEntry<String, DummyConn> entry =
                                f.get(TIMEOUT.getDuration(), TIMEOUT.getTimeUnit());
                        pool.release(entry, true);
                    } catch (final InterruptedException ex) {
                        Thread.currentThread().interrupt();
                    } catch (final ExecutionException | TimeoutException ex) {
                        f.cancel(true);
                        unexpectedException.compareAndSet(null, ex);
                    }
                }));

            }

            for (;;) {
                final Future<?> future = taskQueue.poll();
                if (future != null) {
                    future.cancel(true);
                } else {
                    break;
                }
            }

            executorService.shutdown();
            Assertions.assertTrue(executorService.awaitTermination(TIMEOUT.getDuration(), TIMEOUT.getTimeUnit()));
            Assertions.assertNull(unexpectedException.get());

            final PoolStats stats = pool.getStats(route);
            Assertions.assertEquals(0, stats.getLeased());
        } finally {
            executorService.shutdownNow();
            pool.close(CloseMode.GRACEFUL);
        }
    }