[core] Fix JDBC connection leak in ClientPool and JdbcCatalog

[junmuz]

Purpose

Fixes the Jdbc connection leaks that happen when Flink TMs restart.
Issue: #8267

Tests

There are some tests added specially the testConnectionClosedWhenPoolClosedDuringAction that simulates the scenario.

[junmuz]

@JingsongLi Can I get a review on the PR?

[JingsongLi]

I found one remaining leak race in ClientPool.run. The check at lines 71-72 is not atomic with returning the borrowed client to the deque: a thread can observe this.clients != null, then another thread calls close(), sets this.clients = null and drains the deque, and then the first thread executes clients.addFirst(client) on the old deque. That client was not included in the drain and is no longer reachable from the pool, so it is still left unclosed. This is the same close-during-action case this PR is trying to handle, just with a narrower interleaving. The return-to-pool path and close() need to be made mutually exclusive, or the code needs a post-add closed check/removal that closes the client if the pool was closed concurrently.

[junmuz]

I found one remaining leak race in ClientPool.run. The check at lines 71-72 is not atomic with returning the borrowed client to the deque: a thread can observe this.clients != null, then another thread calls close(), sets this.clients = null and drains the deque, and then the first thread executes clients.addFirst(client) on the old deque. That client was not included in the drain and is no longer reachable from the pool, so it is still left unclosed. This is the same close-during-action case this PR is trying to handle, just with a narrower interleaving. The return-to-pool path and close() need to be made mutually exclusive, or the code needs a post-add closed check/removal that closes the client if the pool was closed concurrently.

@JingsongLi Thanks for the comment. I have tried addressing that.

[nickdelnano]

@junmuz I took a different approach in #8323 to reduce jdbc client connections via sharing per JVM. That approach can reduce connections by a large factor for high parallelism jobs. It would affect some of the logic in this PR as with shared connections we would not explicitly call close.

We would still need the fix of 54a23e4 from this branch, which I have ported there.

Let me know if you have any thoughts.