[Bug] static JdbcConnectionPoolFactory occurs error when multi sources get Jdbc url

[loserwang1024]

Search before asking

• I searched in the issues and found nothing similar.

Flink version

1.17

Flink CDC version

2.4

Database and its version

no Database

Minimal reproduce step

Now JdbcConnectionPoolFactory is often use in JobMaster to reduce connection. In a Job, it's normal to set multiple source, For example, one source from mysql and another source from postgres, and then join them.
Howerver, now as a static variable, JdbcConnectionPoolFactory only have one instance in JobMaster. So only one implements of JdbcConnectionPoolFactory will make sense.The others database will use wrong JdbcConnectionPoolFactory to get jdbc url .
To verify this, only need a simple test:

1. add getJdbcUrl to JdbcConnectionPools for test :

    @VisibleForTesting
    public String getJdbcUrl(JdbcSourceConfig sourceConfig) {
        return jdbcConnectionPoolFactory.getJdbcUrl(sourceConfig);
    }

2. add a test unit:

public class JdbcConnectionPoolFactoryTest {

    @Test
    public void testMultiConnectionPoolFactory() {
        MockConnectionPoolFactory mockConnectionPoolFactory = new MockConnectionPoolFactory();
        MysqlPooledDataSourceFactory mysqlPooledDataSourceFactory = new MysqlPooledDataSourceFactory();
        JdbcConnectionPools mockInstance = JdbcConnectionPools.getInstance(mockConnectionPoolFactory);
        JdbcConnectionPools mysqlInstance = JdbcConnectionPools.getInstance(mysqlPooledDataSourceFactory);
        MySqlSourceConfig mySqlSourceConfig = new MySqlSourceConfig(
                StartupOptions.latest(),
                Arrays.asList("database"),
                Arrays.asList("table"),
                2,
                1,
                1.00,
                2.00,
                false,
                true,
                new Properties(),
                null,
                "com.mysql.cj.jdbc.Driver",
                "localhost",
                8080,
                "username",
                "password",
                20,
                "UTC",
                Duration.ofSeconds(10),
                2,
                3);
        Assert.assertEquals(mockInstance.getJdbcUrl(mySqlSourceConfig), mockConnectionPoolFactory.getJdbcUrl(mySqlSourceConfig));
        //todo: not equal here, because only one JDBCConnectionPoolFactory.
        Assert.assertNotEquals(mysqlInstance.getJdbcUrl(mySqlSourceConfig), mysqlPooledDataSourceFactory.getJdbcUrl(mySqlSourceConfig));
        Assert.assertEquals(mysqlInstance.getJdbcUrl(mySqlSourceConfig), mockConnectionPoolFactory.getJdbcUrl(mySqlSourceConfig));


    }

    static class MockConnectionPoolFactory extends JdbcConnectionPoolFactory{
        @Override
        public String getJdbcUrl(JdbcSourceConfig sourceConfig) {
            return "mock-url";
        }
    }
}

What did you expect to see?

Three assert return true.

What did you see instead?

The later two assert return false

Anything else?

No response

Are you willing to submit a PR?

• I'm willing to submit a PR!

[leonardBang]

👍🏻，pretty detailed analysis