This is an automated email from the ASF dual-hosted git repository.
leonardBang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/master by this push:
new e4aa8f9fe [FLINK-40646][Postgres] Pooled JDBC connections use the
default application name.
e4aa8f9fe is described below
commit e4aa8f9fe548df1b7749cd33566efa490f2cdba3
Author: Hongshun Wang <[email protected]>
AuthorDate: Thu Sep 24 12:12:36 2026 +0800
[FLINK-40646][Postgres] Pooled JDBC connections use the default application
name.
This closes #4538.
---
.../source/PostgresConnectionPoolFactory.java | 9 +++++++--
.../connectors/postgres/source/PostgresDialect.java | 5 +++--
.../cdc/connectors/postgres/PostgresTestBase.java | 3 ++-
.../postgres/source/PostgresDialectTest.java | 19 +++++++++++++++++++
4 files changed, 31 insertions(+), 5 deletions(-)
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresConnectionPoolFactory.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresConnectionPoolFactory.java
index 8a73ac17f..8b996ea98 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresConnectionPoolFactory.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresConnectionPoolFactory.java
@@ -26,7 +26,12 @@ import io.debezium.jdbc.JdbcConfiguration;
/** A connection pool factory to create pooled Postgres {@link
HikariDataSource}. */
public class PostgresConnectionPoolFactory extends JdbcConnectionPoolFactory {
- public static final String JDBC_URL_PATTERN = "jdbc:postgresql://%s:%s/%s";
+ public static final String JDBC_URL_PATTERN =
"jdbc:postgresql://%s:%s/%s?ApplicationName=%s";
+ private final String applicationName;
+
+ public PostgresConnectionPoolFactory(String applicationName) {
+ this.applicationName = applicationName;
+ }
@Override
public String getJdbcUrl(JdbcSourceConfig sourceConfig) {
@@ -34,7 +39,7 @@ public class PostgresConnectionPoolFactory extends
JdbcConnectionPoolFactory {
String hostName = sourceConfig.getHostname();
int port = sourceConfig.getPort();
String database = sourceConfig.getDatabaseList().get(0);
- return String.format(JDBC_URL_PATTERN, hostName, port, database);
+ return String.format(JDBC_URL_PATTERN, hostName, port, database,
applicationName);
}
/**
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresDialect.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresDialect.java
index 474324789..88f7e8446 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresDialect.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresDialect.java
@@ -17,6 +17,7 @@
package org.apache.flink.cdc.connectors.postgres.source;
+import org.apache.flink.cdc.common.annotation.VisibleForTesting;
import org.apache.flink.cdc.connectors.base.config.JdbcSourceConfig;
import org.apache.flink.cdc.connectors.base.dialect.JdbcDataSourceDialect;
import
org.apache.flink.cdc.connectors.base.relational.connection.JdbcConnectionFactory;
@@ -64,7 +65,7 @@ import static
io.debezium.connector.postgresql.Utils.currentOffset;
/** The dialect for Postgres. */
public class PostgresDialect implements JdbcDataSourceDialect {
private static final long serialVersionUID = 1L;
- private static final String CONNECTION_NAME = "postgres-cdc-connector";
+ @VisibleForTesting static final String CONNECTION_NAME =
"postgres-cdc-connector";
private final PostgresSourceConfig sourceConfig;
private transient Tables.TableFilter filters;
@@ -205,7 +206,7 @@ public class PostgresDialect implements
JdbcDataSourceDialect {
@Override
public JdbcConnectionPoolFactory getPooledDataSourceFactory() {
- return new PostgresConnectionPoolFactory();
+ return new PostgresConnectionPoolFactory(CONNECTION_NAME);
}
@Override
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/PostgresTestBase.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/PostgresTestBase.java
index 951b47314..2de9f569e 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/PostgresTestBase.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/PostgresTestBase.java
@@ -125,7 +125,8 @@ public abstract class PostgresTestBase extends
AbstractTestBase {
PostgresConnectionPoolFactory.JDBC_URL_PATTERN,
container.getHost(),
container.getMappedPort(PostgreSQLContainer.POSTGRESQL_PORT),
- databaseName);
+ databaseName,
+ "test-driver");
return DriverManager.getConnection(
jdbcUrl, container.getUsername(), container.getPassword());
}
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/PostgresDialectTest.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/PostgresDialectTest.java
index 0fb9e8007..7a36ef7c9 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/PostgresDialectTest.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/PostgresDialectTest.java
@@ -21,6 +21,7 @@ import
org.apache.flink.cdc.connectors.postgres.PostgresTestBase;
import
org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceConfigFactory;
import org.apache.flink.cdc.connectors.postgres.testutils.UniqueDatabase;
+import io.debezium.connector.postgresql.connection.PostgresConnection;
import io.debezium.relational.TableId;
import org.assertj.core.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -97,6 +98,24 @@ class PostgresDialectTest extends PostgresTestBase {
Assertions.assertThat(tableIdsOfInventoryDatabase2).isEmpty();
}
+ @Test
+ void testJdbcConnectionApplicationName() throws Exception {
+ customDatabase.createAndInitialize();
+ PostgresSourceConfigFactory configFactory =
+ getMockPostgresSourceConfigFactory(customDatabase, "customer",
"Customers", 1);
+ PostgresDialect dialect = new PostgresDialect(configFactory.create(0));
+
+ try (PostgresConnection connection = dialect.openJdbcConnection()) {
+ connection.query(
+ "SHOW application_name",
+ resultSet -> {
+ Assertions.assertThat(resultSet.next()).isTrue();
+ Assertions.assertThat(resultSet.getString(1))
+ .isEqualTo(PostgresDialect.CONNECTION_NAME);
+ });
+ }
+ }
+
@Test
void testDiscoverDataCollectionsForPartitionedTable() {
// initial database with partitioned table
