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 b253248ab7 [Fix][Connector-V2] Clean up stale SFTP pooled sessions
(#11180)
b253248ab7 is described below
commit b253248ab7ba8985ccb9b7f750c01888fb614bfb
Author: Daniel <[email protected]>
AuthorDate: Tue Jun 30 20:06:52 2026 +0800
[Fix][Connector-V2] Clean up stale SFTP pooled sessions (#11180)
---
.../file/sftp/system/SFTPConnectionPool.java | 127 +++++++++-----
.../seatunnel/file/sftp/system/SFTPFileSystem.java | 18 +-
.../file/sftp/system/SFTPConnectionPoolTest.java | 183 +++++++++++++++++++++
3 files changed, 283 insertions(+), 45 deletions(-)
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-sftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sftp/system/SFTPConnectionPool.java
b/seatunnel-connectors-v2/connector-file/connector-file-sftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sftp/system/SFTPConnectionPool.java
index 4906c8061a..677d56c9e9 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-sftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sftp/system/SFTPConnectionPool.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-sftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sftp/system/SFTPConnectionPool.java
@@ -31,6 +31,7 @@ import java.io.IOException;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
+import java.util.Map;
import java.util.Set;
public class SFTPConnectionPool {
@@ -53,6 +54,9 @@ public class SFTPConnectionPool {
}
synchronized ChannelSftp getFromPool(ConnectionInfo info) throws
IOException {
+ if (con2infoMap == null) {
+ throw new IOException("SFTP connection pool has been closed.");
+ }
Set<ChannelSftp> cons = idleConnections.get(info);
ChannelSftp channel;
@@ -60,7 +64,10 @@ public class SFTPConnectionPool {
Iterator<ChannelSftp> it = cons.iterator();
if (it.hasNext()) {
channel = it.next();
- idleConnections.remove(info);
+ it.remove();
+ if (cons.isEmpty()) {
+ idleConnections.remove(info);
+ }
return channel;
} else {
throw new IOException("Connection pool error.");
@@ -69,42 +76,50 @@ public class SFTPConnectionPool {
return null;
}
- synchronized void returnToPool(ChannelSftp channel) {
+ synchronized boolean returnToPool(ChannelSftp channel) {
+ if (con2infoMap == null) {
+ return false;
+ }
ConnectionInfo info = con2infoMap.get(channel);
+ if (info == null) {
+ return false;
+ }
HashSet<ChannelSftp> cons = idleConnections.get(info);
if (cons == null) {
cons = new HashSet<ChannelSftp>();
idleConnections.put(info, cons);
}
cons.add(channel);
+ return true;
}
/** Shutdown the connection pool and close all open connections. */
- synchronized void shutdown() {
- if (this.con2infoMap == null) {
- return; // already shutdown in case it is called
+ void shutdown() {
+ Map<ChannelSftp, ConnectionInfo> connectionsToClose;
+ synchronized (this) {
+ if (this.con2infoMap == null) {
+ return; // already shutdown in case it is called
+ }
+ LOG.info("Inside shutdown, con2infoMap size=" +
con2infoMap.size());
+
+ // Shutdown must close every tracked connection regardless of
live-count drift.
+ connectionsToClose = new HashMap<ChannelSftp,
ConnectionInfo>(con2infoMap);
+ this.maxConnection = 0;
+ this.liveConnectionCount = 0;
+ this.idleConnections = null;
+ this.con2infoMap = null;
}
- LOG.info("Inside shutdown, con2infoMap size=" + con2infoMap.size());
- this.maxConnection = 0;
- Set<ChannelSftp> cons = con2infoMap.keySet();
- if (cons != null && cons.size() > 0) {
- // make a copy since we need to modify the underlying Map
- Set<ChannelSftp> copy = new HashSet<ChannelSftp>(cons);
- // Initiate disconnect from all outstanding connections
- for (ChannelSftp con : copy) {
- try {
- disconnect(con);
- } catch (IOException ioe) {
- ConnectionInfo info = con2infoMap.get(con);
- LOG.error(
- "Error encountered while closing connection to " +
info.getHost(), ioe);
- }
+ for (Map.Entry<ChannelSftp, ConnectionInfo> entry :
connectionsToClose.entrySet()) {
+ try {
+ closeChannel(entry.getKey());
+ } catch (IOException ioe) {
+ LOG.error(
+ "Error encountered while closing connection to "
+ + entry.getValue().getHost(),
+ ioe);
}
}
- // make sure no further connections can be returned.
- this.idleConnections = null;
- this.con2infoMap = null;
}
public synchronized int getMaxConnection() {
@@ -125,11 +140,9 @@ public class SFTPConnectionPool {
if (channel.isConnected()) {
return channel;
} else {
+ removeTrackedChannel(channel);
+ closeChannel(channel);
channel = null;
- synchronized (this) {
- --liveConnectionCount;
- con2infoMap.remove(channel);
- }
}
}
@@ -155,6 +168,9 @@ public class SFTPConnectionPool {
session = jsch.getSession(user, host, port);
}
+ // JSch creates a session reader thread; make it daemon so leaked
sessions cannot keep
+ // Spark local-mode JVMs alive after a batch job has already
finished.
+ session.setDaemonThread(true);
session.setPassword(password);
java.util.Properties config = new java.util.Properties();
@@ -182,31 +198,62 @@ public class SFTPConnectionPool {
// close connection if too many active connections
boolean closeConnection = false;
synchronized (this) {
- if (liveConnectionCount > maxConnection) {
+ if (con2infoMap == null || !con2infoMap.containsKey(channel)) {
+ closeConnection = true;
+ } else if (liveConnectionCount > maxConnection) {
--liveConnectionCount;
con2infoMap.remove(channel);
closeConnection = true;
}
}
if (closeConnection) {
- if (channel.isConnected()) {
- try {
- Session session = channel.getSession();
- channel.disconnect();
- session.disconnect();
- } catch (JSchException e) {
- throw new
IOException(StringUtils.stringifyException(e));
- }
+ closeChannel(channel);
+ } else if (!returnToPool(channel)) {
+ closeChannel(channel);
+ }
+ }
+ }
+
+ /**
+ * Remove bookkeeping for a channel that can no longer be reused before we
create a replacement
+ * connection.
+ */
+ synchronized void removeTrackedChannel(ChannelSftp channel) {
+ if (con2infoMap == null) {
+ return;
+ }
+ ConnectionInfo info = con2infoMap.remove(channel);
+ if (info != null) {
+ Set<ChannelSftp> cons = idleConnections.get(info);
+ if (cons != null) {
+ cons.remove(channel);
+ if (cons.isEmpty()) {
+ idleConnections.remove(info);
}
+ }
+ }
+ if (liveConnectionCount > 0) {
+ liveConnectionCount--;
+ }
+ }
- } else {
- returnToPool(channel);
+ /** Close both the SFTP channel and its backing SSH session when they are
still open. */
+ private void closeChannel(ChannelSftp channel) throws IOException {
+ try {
+ Session session = channel.getSession();
+ if (channel.isConnected()) {
+ channel.disconnect();
}
+ if (session != null && session.isConnected()) {
+ session.disconnect();
+ }
+ } catch (JSchException e) {
+ throw new IOException(StringUtils.stringifyException(e));
}
}
public int getIdleCount() {
- return this.idleConnections.size();
+ return this.idleConnections == null ? 0 : this.idleConnections.size();
}
public int getLiveConnCount() {
@@ -214,7 +261,7 @@ public class SFTPConnectionPool {
}
public int getConnPoolSize() {
- return this.con2infoMap.size();
+ return this.con2infoMap == null ? 0 : this.con2infoMap.size();
}
/**
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-sftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sftp/system/SFTPFileSystem.java
b/seatunnel-connectors-v2/connector-file/connector-file-sftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sftp/system/SFTPFileSystem.java
index 99bf417763..0990726aa0 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-sftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sftp/system/SFTPFileSystem.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-sftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sftp/system/SFTPFileSystem.java
@@ -115,7 +115,7 @@ public class SFTPFileSystem extends FileSystem {
}
int connectionMax = conf.getInt(FS_SFTP_CONNECTION_MAX,
DEFAULT_MAX_CONNECTION);
- connectionPool = new SFTPConnectionPool(connectionMax, connectionMax);
+ connectionPool = new SFTPConnectionPool(connectionMax, 0);
}
private ChannelSftp connect() throws IOException {
@@ -548,8 +548,11 @@ public class SFTPFileSystem extends FileSystem {
new FSDataOutputStream(os, statistics) {
@Override
public void close() throws IOException {
- super.close();
- disconnect(client);
+ try {
+ super.close();
+ } finally {
+ disconnect(client);
+ }
}
};
@@ -651,7 +654,12 @@ public class SFTPFileSystem extends FileSystem {
@Override
public void close() throws IOException {
- super.close();
- connectionPool.shutdown();
+ try {
+ super.close();
+ } finally {
+ if (connectionPool != null) {
+ connectionPool.shutdown();
+ }
+ }
}
}
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-sftp/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/sftp/system/SFTPConnectionPoolTest.java
b/seatunnel-connectors-v2/connector-file/connector-file-sftp/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/sftp/system/SFTPConnectionPoolTest.java
new file mode 100644
index 0000000000..cf26cbfbe5
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-file/connector-file-sftp/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/sftp/system/SFTPConnectionPoolTest.java
@@ -0,0 +1,183 @@
+/*
+ * 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.file.sftp.system;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import com.jcraft.jsch.ChannelSftp;
+import com.jcraft.jsch.JSchException;
+import com.jcraft.jsch.Session;
+
+import java.lang.reflect.Field;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * Tests the SFTP connection pool bookkeeping that continuous discovery relies
on during cleanup.
+ */
+class SFTPConnectionPoolTest {
+
+ /**
+ * Keep sibling idle channels tracked after one channel is borrowed from
the shared pool key.
+ */
+ @Test
+ void getFromPoolShouldKeepOtherIdleChannelsTracked() throws Exception {
+ SFTPConnectionPool connectionPool = new SFTPConnectionPool(2, 0);
+ SFTPConnectionPool.ConnectionInfo connectionInfo =
+ new SFTPConnectionPool.ConnectionInfo("host", 22, "user");
+ ChannelSftp firstChannel = new TestChannelSftp(true,
Mockito.mock(Session.class));
+ ChannelSftp secondChannel = new TestChannelSftp(true,
Mockito.mock(Session.class));
+
+ HashSet<ChannelSftp> idleChannels = new HashSet<>();
+ idleChannels.add(firstChannel);
+ idleChannels.add(secondChannel);
+ idleConnections(connectionPool).put(connectionInfo, idleChannels);
+ trackedConnections(connectionPool).put(firstChannel, connectionInfo);
+ trackedConnections(connectionPool).put(secondChannel, connectionInfo);
+
+ ChannelSftp borrowedChannel =
connectionPool.getFromPool(connectionInfo);
+
+ Assertions.assertTrue(borrowedChannel == firstChannel ||
borrowedChannel == secondChannel);
+ Set<ChannelSftp> remainingIdleChannels =
+ idleConnections(connectionPool).get(connectionInfo);
+ Assertions.assertNotNull(remainingIdleChannels);
+ Assertions.assertEquals(1, remainingIdleChannels.size());
+
Assertions.assertFalse(remainingIdleChannels.contains(borrowedChannel));
+ }
+
+ /** Always disconnect the SSH session when a stale channel is being
permanently closed. */
+ @Test
+ void disconnectShouldCloseSessionForDisconnectedChannel() throws Exception
{
+ SFTPConnectionPool connectionPool = new SFTPConnectionPool(0, 1);
+ SFTPConnectionPool.ConnectionInfo connectionInfo =
+ new SFTPConnectionPool.ConnectionInfo("host", 22, "user");
+ Session session = Mockito.mock(Session.class);
+ Mockito.when(session.isConnected()).thenReturn(true);
+ TestChannelSftp channel = new TestChannelSftp(false, session);
+ trackedConnections(connectionPool).put(channel, connectionInfo);
+
+ connectionPool.disconnect(channel);
+
+ Mockito.verify(session).disconnect();
+ Assertions.assertFalse(channel.wasDisconnected());
+ Assertions.assertEquals(0, connectionPool.getLiveConnCount());
+ Assertions.assertEquals(0, connectionPool.getConnPoolSize());
+ }
+
+ /** Shutdown must close tracked sessions even when live-count bookkeeping
has drifted. */
+ @Test
+ void shutdownShouldCloseTrackedChannelsRegardlessOfLiveCount() throws
Exception {
+ SFTPConnectionPool connectionPool = new SFTPConnectionPool(2, 0);
+ SFTPConnectionPool.ConnectionInfo connectionInfo =
+ new SFTPConnectionPool.ConnectionInfo("host", 22, "user");
+ Session session = Mockito.mock(Session.class);
+ Mockito.when(session.isConnected()).thenReturn(true);
+ TestChannelSftp channel = new TestChannelSftp(true, session);
+ trackedConnections(connectionPool).put(channel, connectionInfo);
+
+ connectionPool.shutdown();
+
+ Assertions.assertTrue(channel.wasDisconnected());
+ Mockito.verify(session).disconnect();
+ }
+
+ /** Late disconnects after shutdown must close the channel instead of
touching closed maps. */
+ @Test
+ void disconnectAfterShutdownShouldCloseChannel() throws Exception {
+ SFTPConnectionPool connectionPool = new SFTPConnectionPool(1, 0);
+ Session session = Mockito.mock(Session.class);
+ Mockito.when(session.isConnected()).thenReturn(true);
+ TestChannelSftp channel = new TestChannelSftp(true, session);
+
+ connectionPool.shutdown();
+ connectionPool.disconnect(channel);
+
+ Assertions.assertTrue(channel.wasDisconnected());
+ Mockito.verify(session).disconnect();
+ Assertions.assertEquals(0, connectionPool.getIdleCount());
+ Assertions.assertEquals(0, connectionPool.getConnPoolSize());
+ }
+
+ /** Unknown channels are not reusable pool entries, so they must be closed
immediately. */
+ @Test
+ void disconnectShouldCloseUntrackedChannel() throws Exception {
+ SFTPConnectionPool connectionPool = new SFTPConnectionPool(1, 1);
+ Session session = Mockito.mock(Session.class);
+ Mockito.when(session.isConnected()).thenReturn(true);
+ TestChannelSftp channel = new TestChannelSftp(true, session);
+
+ connectionPool.disconnect(channel);
+
+ Assertions.assertTrue(channel.wasDisconnected());
+ Mockito.verify(session).disconnect();
+ Assertions.assertEquals(0, connectionPool.getIdleCount());
+ }
+
+ /** Small concrete ChannelSftp stub that keeps Mockito away from final
JSch internals. */
+ private static final class TestChannelSftp extends ChannelSftp {
+ private final boolean connected;
+ private final Session session;
+ private boolean disconnected;
+
+ private TestChannelSftp(boolean connected, Session session) {
+ this.connected = connected;
+ this.session = session;
+ }
+
+ @Override
+ public boolean isConnected() {
+ return connected;
+ }
+
+ @Override
+ public void disconnect() {
+ disconnected = true;
+ }
+
+ @Override
+ public Session getSession() throws JSchException {
+ return session;
+ }
+
+ private boolean wasDisconnected() {
+ return disconnected;
+ }
+ }
+
+ /** Read the private idle map to assert that the pool does not lose
sibling channels. */
+ @SuppressWarnings("unchecked")
+ private static Map<SFTPConnectionPool.ConnectionInfo,
HashSet<ChannelSftp>> idleConnections(
+ SFTPConnectionPool connectionPool) throws Exception {
+ Field field =
SFTPConnectionPool.class.getDeclaredField("idleConnections");
+ field.setAccessible(true);
+ return (Map<SFTPConnectionPool.ConnectionInfo, HashSet<ChannelSftp>>)
+ field.get(connectionPool);
+ }
+
+ /** Read the private tracked-channel map to seed deterministic pool state
for unit tests. */
+ @SuppressWarnings("unchecked")
+ private static Map<ChannelSftp, SFTPConnectionPool.ConnectionInfo>
trackedConnections(
+ SFTPConnectionPool connectionPool) throws Exception {
+ Field field = SFTPConnectionPool.class.getDeclaredField("con2infoMap");
+ field.setAccessible(true);
+ return (Map<ChannelSftp, SFTPConnectionPool.ConnectionInfo>)
field.get(connectionPool);
+ }
+}
