[mysql] Connection pool id should contains username (#1328)

flink-cdc-base/src/main/java/com/ververica/cdc/connectors/base/relational/connection/ConnectionPoolId.java

@@ -27,10 +27,12 @@ public class ConnectionPoolId implements Serializable {
     private static final long serialVersionUID = 1L;
     private final String host;
     private final int port;
+    private final String username;
 
-    public ConnectionPoolId(String host, int port) {
+    public ConnectionPoolId(String host, int port, String username) {
         this.host = host;
         this.port = port;
+        this.username = username;
     }
 
     @Override
@@ -42,16 +44,18 @@ public class ConnectionPoolId implements Serializable {
             return false;
         }
         ConnectionPoolId that = (ConnectionPoolId) o;
-        return Objects.equals(host, that.host) && Objects.equals(port, that.port);
+        return Objects.equals(host, that.host)
+                && Objects.equals(port, that.port)
+                && Objects.equals(username, that.username);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(host, port);
+        return Objects.hash(host, port, username);
     }
 
     @Override
     public String toString() {
-        return host + ':' + port;
+        return username + '@' + host + ':' + port;
     }
 }

flink-cdc-base/src/main/java/com/ververica/cdc/connectors/base/relational/connection/JdbcConnectionFactory.java

@@ -49,7 +49,10 @@ public class JdbcConnectionFactory implements JdbcConnection.ConnectionFactory {
         final int connectRetryTimes = sourceConfig.getConnectMaxRetries();
 
         final ConnectionPoolId connectionPoolId =
-                new ConnectionPoolId(sourceConfig.getHostname(), sourceConfig.getPort());
+                new ConnectionPoolId(
+                        sourceConfig.getHostname(),
+                        sourceConfig.getPort(),
+                        sourceConfig.getUsername());
 
         HikariDataSource dataSource =
                 JdbcConnectionPools.getInstance(jdbcConnectionPoolFactory)

flink-connector-mysql-cdc/src/main/java/com/ververica/cdc/connectors/mysql/source/connection/ConnectionPoolId.java

@@ -27,10 +27,12 @@ public class ConnectionPoolId implements Serializable {
     private static final long serialVersionUID = 1L;
     private final String host;
     private final int port;
+    private final String username;
 
-    public ConnectionPoolId(String host, int port) {
+    public ConnectionPoolId(String host, int port, String username) {
         this.host = host;
         this.port = port;
+        this.username = username;
     }
 
     @Override
@@ -42,16 +44,18 @@ public class ConnectionPoolId implements Serializable {
             return false;
         }
         ConnectionPoolId that = (ConnectionPoolId) o;
-        return Objects.equals(host, that.host) && Objects.equals(port, that.port);
+        return Objects.equals(host, that.host)
+                && Objects.equals(port, that.port)
+                && Objects.equals(username, that.username);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(host, port);
+        return Objects.hash(host, port, username);
     }
 
     @Override
     public String toString() {
-        return host + ':' + port;
+        return username + '@' + host + ':' + port;
     }
 }

flink-connector-mysql-cdc/src/main/java/com/ververica/cdc/connectors/mysql/source/connection/JdbcConnectionFactory.java

@@ -46,7 +46,10 @@ public class JdbcConnectionFactory implements JdbcConnection.ConnectionFactory {
         final int connectRetryTimes = sourceConfig.getConnectMaxRetries();
 
         final ConnectionPoolId connectionPoolId =
-                new ConnectionPoolId(sourceConfig.getHostname(), sourceConfig.getPort());
+                new ConnectionPoolId(
+                        sourceConfig.getHostname(),
+                        sourceConfig.getPort(),
+                        sourceConfig.getUsername());
 
         HikariDataSource dataSource =
                 JdbcConnectionPools.getInstance()