This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new b8d5405f44 [Fix][Connector-V2][Jdbc] Validate a cached connection
before reusing it (#11678)
b8d5405f44 is described below
commit b8d5405f440b685f842546bd2feb98a37eb3bcf7
Author: Santhosh Kumar Somarapu <[email protected]>
AuthorDate: Sat Aug 22 19:57:36 2026 -0500
[Fix][Connector-V2][Jdbc] Validate a cached connection before reusing it
(#11678)
Signed-off-by: Santhosh Kumar Somarapu <[email protected]>
---
.../SimpleJdbcConnectionPoolProviderProxy.java | 7 +
.../seatunnel/jdbc/sink/ConnectionPoolManager.java | 114 +++++++++-
.../seatunnel/jdbc/sink/JdbcSinkWriter.java | 41 ++--
.../jdbc/sink/ConnectionPoolManagerTest.java | 230 +++++++++++++++++++++
4 files changed, 370 insertions(+), 22 deletions(-)
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/connection/SimpleJdbcConnectionPoolProviderProxy.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/connection/SimpleJdbcConnectionPoolProviderProxy.java
index 6a3f024e56..f9c83318b0 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/connection/SimpleJdbcConnectionPoolProviderProxy.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/connection/SimpleJdbcConnectionPoolProviderProxy.java
@@ -44,6 +44,13 @@ public class SimpleJdbcConnectionPoolProviderProxy
implements JdbcConnectionProv
return poolManager.getConnection(queueIndex);
}
+ /**
+ * Note that this is not a pure predicate. {@code getConnection} validates
the cached connection
+ * and replaces it when it is no longer usable, so calling this can repair
the cached connection
+ * as a side effect and then report it valid. That is intended: for a
connection held per queue
+ * index there is nothing a caller could usefully do with "invalid" other
than ask for a new
+ * one.
+ */
@Override
public boolean isConnectionValid() throws SQLException {
return poolManager.containsConnection(queueIndex)
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/sink/ConnectionPoolManager.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/sink/ConnectionPoolManager.java
index 67778b945b..08a49c4c3f 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/sink/ConnectionPoolManager.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/sink/ConnectionPoolManager.java
@@ -20,28 +20,61 @@ package org.apache.seatunnel.connectors.seatunnel.jdbc.sink;
import org.apache.seatunnel.shade.com.zaxxer.hikari.HikariDataSource;
import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
import java.sql.Connection;
import java.sql.SQLException;
+import java.sql.Statement;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+@Slf4j
@Getter
public class ConnectionPoolManager {
+ private static final long DEFAULT_VALIDATION_TIMEOUT_MILLIS = 5000L;
+
+ private static final long REPLACEMENT_WARN_INTERVAL_NANOS =
TimeUnit.MINUTES.toNanos(1);
+
private final HikariDataSource connectionPool;
private final Map<Integer, Connection> connectionMap;
+ private final AtomicLong replacementsSinceLastWarn = new AtomicLong();
+
+ private final AtomicLong lastReplacementWarnNanos;
+
ConnectionPoolManager(HikariDataSource connectionPool) {
this.connectionPool = connectionPool;
connectionMap = new ConcurrentHashMap<>();
+ // Backdated so the first replacement warns immediately rather than
waiting out an interval.
+ lastReplacementWarnNanos =
+ new AtomicLong(System.nanoTime() -
REPLACEMENT_WARN_INTERVAL_NANOS);
}
+ /**
+ * Returns the connection held for this queue index, replacing it first if
it is no longer
+ * usable.
+ *
+ * <p>A connection is borrowed from the pool once per index and then held
for the lifetime of
+ * the writer, so it never goes back for Hikari's housekeeper to inspect.
On a streaming job the
+ * gap between writes can be hours, and the server or anything on the path
can close the socket
+ * in the meantime. Handing the cached connection back without checking it
means the failure
+ * surfaces as a write error after the record has already been consumed.
+ */
public Connection getConnection(int index) {
- return connectionMap.computeIfAbsent(
+ return connectionMap.compute(
index,
- i -> {
+ (i, cached) -> {
+ if (cached != null && isUsable(cached)) {
+ return cached;
+ }
+ if (cached != null) {
+ logReplacement(i);
+ }
+ closeQuietly(cached);
try {
return connectionPool.getConnection();
} catch (SQLException e) {
@@ -50,6 +83,83 @@ public class ConnectionPoolManager {
});
}
+ /**
+ * Uses the pool's own validation settings rather than a second opinion:
{@code
+ * connectionTestQuery} when one is configured, because it is configured
precisely for drivers
+ * whose {@code isValid} cannot be trusted, and {@code isValid} otherwise.
+ */
+ private boolean isUsable(Connection connection) {
+ try {
+ if (connection.isClosed()) {
+ return false;
+ }
+ int timeoutSeconds =
+ (int)
+ Math.max(
+ 1,
+
TimeUnit.MILLISECONDS.toSeconds(getValidationTimeoutMillis()));
+ String testQuery = connectionPool.getConnectionTestQuery();
+ if (testQuery == null) {
+ return connection.isValid(timeoutSeconds);
+ }
+ try (Statement statement = connection.createStatement()) {
+ statement.setQueryTimeout(timeoutSeconds);
+ statement.execute(testQuery);
+ return true;
+ }
+ } catch (SQLException e) {
+ log.debug("Cached connection for index is no longer usable,
replacing it", e);
+ return false;
+ }
+ }
+
+ /**
+ * Reports a replaced connection at WARN, rate limited to one message per
minute per manager.
+ *
+ * <p>Replacing an occasionally idle connection is the expected case and
is not worth a warning
+ * on every occurrence. A validation that fails systematically is not: a
misconfigured {@code
+ * connectionTestQuery}, a persistent network fault or a server-side
connection limit makes
+ * every call replace the connection, which silently turns the cache off
and churns connections
+ * continuously. Logging only at DEBUG would leave that indistinguishable
from healthy operation
+ * in a default deployment, so the first occurrence and a periodic summary
are surfaced.
+ */
+ private void logReplacement(int index) {
+ long replacements = replacementsSinceLastWarn.incrementAndGet();
+ long now = System.nanoTime();
+ long last = lastReplacementWarnNanos.get();
+
+ if (now - last >= REPLACEMENT_WARN_INTERVAL_NANOS
+ && lastReplacementWarnNanos.compareAndSet(last, now)) {
+ replacementsSinceLastWarn.addAndGet(-replacements);
+ log.warn(
+ "Replaced an unusable pooled connection for queue index
{}. "
+ + "{} replacement(s) since the last such message.
Repeated messages "
+ + "indicate the connection is not surviving
between writes.",
+ index,
+ replacements);
+
+ return;
+ }
+
+ log.debug("Cached connection for queue index {} is no longer usable,
replacing it", index);
+ }
+
+ private long getValidationTimeoutMillis() {
+ long configured = connectionPool.getValidationTimeout();
+ return configured > 0 ? configured : DEFAULT_VALIDATION_TIMEOUT_MILLIS;
+ }
+
+ private void closeQuietly(Connection connection) {
+ if (connection == null) {
+ return;
+ }
+ try {
+ connection.close();
+ } catch (SQLException e) {
+ log.debug("Failed to close an unusable connection, discarding it
anyway", e);
+ }
+ }
+
public boolean containsConnection(int index) {
return connectionMap.containsKey(index);
}
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/sink/JdbcSinkWriter.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/sink/JdbcSinkWriter.java
index 18a5f85118..36cad4e525 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/sink/JdbcSinkWriter.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/sink/JdbcSinkWriter.java
@@ -46,6 +46,7 @@ import
org.apache.seatunnel.connectors.seatunnel.jdbc.state.XidInfo;
import lombok.extern.slf4j.Slf4j;
import java.io.IOException;
+import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.SQLException;
import java.sql.Savepoint;
@@ -354,8 +355,9 @@ public class JdbcSinkWriter extends
AbstractJdbcSinkWriter<ConnectionPoolManager
tryOpen();
outputFormat.flush();
try {
- if (!connectionProvider.getConnection().getAutoCommit()) {
- connectionProvider.getConnection().commit();
+ Connection connection = connectionProvider.getConnection();
+ if (!connection.getAutoCommit()) {
+ connection.commit();
}
} catch (SQLException e) {
throw new JdbcConnectorException(
@@ -388,9 +390,10 @@ public class JdbcSinkWriter extends
AbstractJdbcSinkWriter<ConnectionPoolManager
return false;
}
try {
+ Connection connection = connectionProvider.getConnection();
Savepoint previousSavepoint = lastSuccessfulBatchSavepoint;
- lastSuccessfulBatchSavepoint =
connectionProvider.getConnection().setSavepoint();
- releaseSavepointSilently(previousSavepoint);
+ lastSuccessfulBatchSavepoint = connection.setSavepoint();
+ releaseSavepointSilently(connection, previousSavepoint);
return true;
} catch (SQLException e) {
throw new JdbcConnectorException(
@@ -430,12 +433,12 @@ public class JdbcSinkWriter extends
AbstractJdbcSinkWriter<ConnectionPoolManager
// Releasing an old savepoint is a best-effort cleanup. Some drivers
invalidate savepoints after
// rollback/commit and should not fail the writer just because cleanup is
no longer possible.
- private void releaseSavepointSilently(Savepoint savepoint) {
+ private void releaseSavepointSilently(Connection connection, Savepoint
savepoint) {
if (savepoint == null) {
return;
}
try {
- connectionProvider.getConnection().releaseSavepoint(savepoint);
+ connection.releaseSavepoint(savepoint);
} catch (SQLException e) {
log.debug("Failed to release JDBC savepoint after moving row-error
batch boundary.", e);
}
@@ -491,24 +494,22 @@ public class JdbcSinkWriter extends
AbstractJdbcSinkWriter<ConnectionPoolManager
}
private void rollbackIfNeeded() throws SQLException {
- try {
- if (connectionProvider.getConnection().getAutoCommit()) {
- return;
- }
- if (lastSuccessfulBatchSavepoint != null) {
-
connectionProvider.getConnection().rollback(lastSuccessfulBatchSavepoint);
- } else {
- connectionProvider.getConnection().rollback();
- }
- lastSuccessfulBatchSavepoint = null;
- } catch (SQLException rollbackEx) {
- throw rollbackEx;
+ Connection connection = connectionProvider.getConnection();
+ if (connection.getAutoCommit()) {
+ return;
+ }
+ if (lastSuccessfulBatchSavepoint != null) {
+ connection.rollback(lastSuccessfulBatchSavepoint);
+ } else {
+ connection.rollback();
}
+ lastSuccessfulBatchSavepoint = null;
}
private void commitIfNeeded() throws SQLException {
- if (!connectionProvider.getConnection().getAutoCommit()) {
- connectionProvider.getConnection().commit();
+ Connection connection = connectionProvider.getConnection();
+ if (!connection.getAutoCommit()) {
+ connection.commit();
lastSuccessfulBatchSavepoint = null;
}
}
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/sink/ConnectionPoolManagerTest.java
b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/sink/ConnectionPoolManagerTest.java
new file mode 100644
index 0000000000..f0734de246
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/sink/ConnectionPoolManagerTest.java
@@ -0,0 +1,230 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.jdbc.sink;
+
+import org.apache.seatunnel.shade.com.zaxxer.hikari.HikariDataSource;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.sql.Statement;
+
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class ConnectionPoolManagerTest {
+
+ @Test
+ void reusesAConnectionThatIsStillUsable() throws SQLException {
+ Connection live = mock(Connection.class);
+ when(live.isClosed()).thenReturn(false);
+ when(live.isValid(anyInt())).thenReturn(true);
+
+ HikariDataSource pool = mock(HikariDataSource.class);
+ when(pool.getConnection()).thenReturn(live);
+
+ ConnectionPoolManager manager = new ConnectionPoolManager(pool);
+
+ Assertions.assertSame(live, manager.getConnection(0));
+ Assertions.assertSame(live, manager.getConnection(0));
+
+ // Borrowed once, then served from the cache.
+ verify(pool, times(1)).getConnection();
+ verify(live, never()).close();
+ }
+
+ @Test
+ void replacesAConnectionThatHasGoneStale() throws SQLException {
+ // A streaming sink can sit idle for hours between writes. The server
closes the socket in
+ // the meantime, and the cached connection is dead by the time the
next record arrives.
+ Connection stale = mock(Connection.class);
+ when(stale.isClosed()).thenReturn(false);
+ when(stale.isValid(anyInt())).thenReturn(false);
+
+ Connection fresh = mock(Connection.class);
+ when(fresh.isClosed()).thenReturn(false);
+ when(fresh.isValid(anyInt())).thenReturn(true);
+
+ HikariDataSource pool = mock(HikariDataSource.class);
+ when(pool.getConnection()).thenReturn(stale, fresh);
+
+ ConnectionPoolManager manager = new ConnectionPoolManager(pool);
+
+ Assertions.assertSame(stale, manager.getConnection(0));
+ Assertions.assertSame(fresh, manager.getConnection(0));
+
+ verify(pool, times(2)).getConnection();
+ // The dead one is handed back rather than leaked.
+ verify(stale, times(1)).close();
+ }
+
+ @Test
+ void replacesAConnectionThatIsAlreadyClosed() throws SQLException {
+ Connection closed = mock(Connection.class);
+ when(closed.isClosed()).thenReturn(true);
+
+ Connection fresh = mock(Connection.class);
+ when(fresh.isClosed()).thenReturn(false);
+ when(fresh.isValid(anyInt())).thenReturn(true);
+
+ HikariDataSource pool = mock(HikariDataSource.class);
+ when(pool.getConnection()).thenReturn(closed, fresh);
+
+ ConnectionPoolManager manager = new ConnectionPoolManager(pool);
+
+ Assertions.assertSame(closed, manager.getConnection(0));
+ Assertions.assertSame(fresh, manager.getConnection(0));
+
+ // isValid is never consulted once isClosed has answered.
+ verify(closed, never()).isValid(anyInt());
+ }
+
+ @Test
+ void usesTheConfiguredTestQueryInsteadOfIsValid() throws SQLException {
+ // connectionTestQuery is set for drivers whose isValid cannot be
trusted, so the check has
+ // to honour it rather than form a second opinion.
+ Statement statement = mock(Statement.class);
+ Connection connection = mock(Connection.class);
+ when(connection.isClosed()).thenReturn(false);
+ when(connection.createStatement()).thenReturn(statement);
+
+ HikariDataSource pool = mock(HikariDataSource.class);
+ when(pool.getConnection()).thenReturn(connection);
+ when(pool.getConnectionTestQuery()).thenReturn("SELECT 1");
+
+ ConnectionPoolManager manager = new ConnectionPoolManager(pool);
+
+ Assertions.assertSame(connection, manager.getConnection(0));
+ Assertions.assertSame(connection, manager.getConnection(0));
+
+ verify(statement, times(1)).execute("SELECT 1");
+ verify(connection, never()).isValid(anyInt());
+ }
+
+ @Test
+ void replacesTheConnectionWhenTheTestQueryFails() throws SQLException {
+ Statement failing = mock(Statement.class);
+ when(failing.execute(anyString())).thenThrow(new
SQLException("connection is closed"));
+
+ Connection stale = mock(Connection.class);
+ when(stale.isClosed()).thenReturn(false);
+ when(stale.createStatement()).thenReturn(failing);
+
+ Connection fresh = mock(Connection.class);
+ when(fresh.isClosed()).thenReturn(false);
+ when(fresh.createStatement()).thenReturn(mock(Statement.class));
+
+ HikariDataSource pool = mock(HikariDataSource.class);
+ when(pool.getConnection()).thenReturn(stale, fresh);
+ when(pool.getConnectionTestQuery()).thenReturn("SELECT 1");
+
+ ConnectionPoolManager manager = new ConnectionPoolManager(pool);
+
+ Assertions.assertSame(stale, manager.getConnection(0));
+ Assertions.assertSame(fresh, manager.getConnection(0));
+
+ verify(pool, times(2)).getConnection();
+ }
+
+ @Test
+ void passesThePoolsValidationTimeoutToIsValid() throws SQLException {
+ Connection connection = mock(Connection.class);
+ when(connection.isClosed()).thenReturn(false);
+ when(connection.isValid(anyInt())).thenReturn(true);
+
+ HikariDataSource pool = mock(HikariDataSource.class);
+ when(pool.getConnection()).thenReturn(connection);
+ when(pool.getValidationTimeout()).thenReturn(8000L);
+
+ ConnectionPoolManager manager = new ConnectionPoolManager(pool);
+
+ manager.getConnection(0);
+ manager.getConnection(0);
+
+ // Seconds, rounded down from the pool's own millisecond setting.
+ verify(connection, times(1)).isValid(8);
+ }
+
+ @Test
+ void fallsBackToFiveSecondsWhenThePoolHasNoValidationTimeout() throws
SQLException {
+ Connection connection = mock(Connection.class);
+ when(connection.isClosed()).thenReturn(false);
+ when(connection.isValid(anyInt())).thenReturn(true);
+
+ HikariDataSource pool = mock(HikariDataSource.class);
+ when(pool.getConnection()).thenReturn(connection);
+ // Hikari reports 0 when the validation timeout has not been
configured.
+ when(pool.getValidationTimeout()).thenReturn(0L);
+
+ ConnectionPoolManager manager = new ConnectionPoolManager(pool);
+
+ manager.getConnection(0);
+ manager.getConnection(0);
+
+ verify(connection, times(1)).isValid(5);
+ }
+
+ @Test
+ void neverPassesATimeoutBelowOneSecond() throws SQLException {
+ Connection connection = mock(Connection.class);
+ when(connection.isClosed()).thenReturn(false);
+ when(connection.isValid(anyInt())).thenReturn(true);
+
+ HikariDataSource pool = mock(HikariDataSource.class);
+ when(pool.getConnection()).thenReturn(connection);
+ // Sub-second timeouts truncate to zero seconds, which isValid treats
as no timeout at all.
+ when(pool.getValidationTimeout()).thenReturn(250L);
+
+ ConnectionPoolManager manager = new ConnectionPoolManager(pool);
+
+ manager.getConnection(0);
+ manager.getConnection(0);
+
+ verify(connection, times(1)).isValid(1);
+ }
+
+ @Test
+ void keepsConnectionsSeparatePerQueueIndex() throws SQLException {
+ Connection first = mock(Connection.class);
+ when(first.isClosed()).thenReturn(false);
+ when(first.isValid(anyInt())).thenReturn(true);
+
+ Connection second = mock(Connection.class);
+ when(second.isClosed()).thenReturn(false);
+ when(second.isValid(anyInt())).thenReturn(true);
+
+ HikariDataSource pool = mock(HikariDataSource.class);
+ when(pool.getConnection()).thenReturn(first, second);
+
+ ConnectionPoolManager manager = new ConnectionPoolManager(pool);
+
+ Assertions.assertSame(first, manager.getConnection(0));
+ Assertions.assertSame(second, manager.getConnection(1));
+ Assertions.assertSame(first, manager.getConnection(0));
+ Assertions.assertSame(second, manager.getConnection(1));
+
+ verify(pool, times(2)).getConnection();
+ }
+}
