This is an automated email from the ASF dual-hosted git repository.
sruehl pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/plc4x.git
The following commit(s) were added to refs/heads/develop by this push:
new 130d77d37b fix(plc4j/transports-tcp): close SocketChannel on failed
bind/socket-option setup (#2645)
130d77d37b is described below
commit 130d77d37b647d8e6426463de7a61198d9b33471
Author: Martin Kribs <[email protected]>
AuthorDate: Thu Jul 23 13:59:10 2026 +0200
fix(plc4j/transports-tcp): close SocketChannel on failed bind/socket-option
setup (#2645)
* fix(plc4j/transports-tcp): close SocketChannel on failed
bind/socket-option setup
TcpTransportInstance's constructor only assigned the opened SocketChannel
to the socketChannel field after bind(), socket options and connect() all
succeeded. If bind() or a socket-option call threw (e.g. a pinned local
port still held by a previous failed attempt), the already-open channel
was never closed - each such failure leaked one fd. connect() failures
(timeout/refused) are unaffected since Socket.connect() already closes
itself internally on failure.
* fix(plc4j/transports-tcp): tighten fd-leak test assertion per review
grown < attempts still passed if the leak only fired on some (but not all)
iterations. Assert grown <= 1 instead, so any per-attempt leak is caught, not
just a leak on every single attempt.
Co-authored-by: Copilot Autofix powered by AI
<[email protected]>
* fix(plc4j/transports-tcp): address Copilot review feedback on fd-leak test
* fix(plc4j/transports-tcp): close channel on any constructor failure, not
just IOException
---------
Co-authored-by: Copilot Autofix powered by AI
<[email protected]>
---
.../java/transport/tcp/TcpTransportInstance.java | 36 ++++++++++------
.../transport/tcp/TcpTransportInstanceTest.java | 49 ++++++++++++++++++++++
2 files changed, 73 insertions(+), 12 deletions(-)
diff --git
a/plc4j/transports/tcp/src/main/java/org/apache/plc4x/java/transport/tcp/TcpTransportInstance.java
b/plc4j/transports/tcp/src/main/java/org/apache/plc4x/java/transport/tcp/TcpTransportInstance.java
index 6b67613c05..f6325c50ca 100644
---
a/plc4j/transports/tcp/src/main/java/org/apache/plc4x/java/transport/tcp/TcpTransportInstance.java
+++
b/plc4j/transports/tcp/src/main/java/org/apache/plc4x/java/transport/tcp/TcpTransportInstance.java
@@ -74,26 +74,29 @@ public class TcpTransportInstance extends
BaseTransportInstance<TcpTransportConf
this.ringBuffer = new RingBuffer(configuration.receiveBufferSize);
this.readBuffer = ByteBuffer.allocateDirect(DEFAULT_BUFFER_SIZE); //
Reused direct buffer for channel reads
+ // Opened into a local first, only promoted to the final
`socketChannel` field once every
+ // setup step (bind, socket options, connect) has succeeded.
+ SocketChannel channel = null;
try {
// Open socket channel
- this.socketChannel = SocketChannel.open();
+ channel = SocketChannel.open();
// Bind to a local address if specified
if (configuration.localAddress != null &&
!configuration.localAddress.isEmpty()) {
SocketAddress localAddr = new
InetSocketAddress(configuration.localAddress, configuration.localPort);
- socketChannel.bind(localAddr);
+ channel.bind(localAddr);
LOGGER.debug("Bound to local address {}:{}",
configuration.localAddress, configuration.localPort);
}
// Configure socket options before connecting
- socketChannel.socket().setTcpNoDelay(configuration.tcpNoDelay);
- socketChannel.socket().setKeepAlive(configuration.keepAlive);
+ channel.socket().setTcpNoDelay(configuration.tcpNoDelay);
+ channel.socket().setKeepAlive(configuration.keepAlive);
if (configuration.sendBufferSize > 0) {
-
socketChannel.socket().setSendBufferSize(configuration.sendBufferSize);
+
channel.socket().setSendBufferSize(configuration.sendBufferSize);
}
if (configuration.receiveBufferSize > 0) {
-
socketChannel.socket().setReceiveBufferSize(configuration.receiveBufferSize);
+
channel.socket().setReceiveBufferSize(configuration.receiveBufferSize);
}
// Note: configuration.readTimeout is intentionally NOT mapped to
Socket.setSoTimeout here.
// SO_TIMEOUT has no effect on blocking SocketChannel reads, so
the previous call was a
@@ -101,11 +104,13 @@ public class TcpTransportInstance extends
BaseTransportInstance<TcpTransportConf
// bounds responses via CompletableFuture timeouts), not by the
transport.
// Connect with timeout
- socketChannel.socket().connect(remoteAddress,
configuration.connectTimeout);
+ channel.socket().connect(remoteAddress,
configuration.connectTimeout);
// Blocking mode: on Java 21 a virtual thread blocked in
read()/write() parks and
// releases its carrier, so no selector is needed.
- socketChannel.configureBlocking(true);
+ channel.configureBlocking(true);
+
+ this.socketChannel = channel;
LOGGER.info("Connected to {}:{} with async support",
remoteAddress.getHostName(), remoteAddress.getPort());
@@ -114,13 +119,20 @@ public class TcpTransportInstance extends
BaseTransportInstance<TcpTransportConf
remoteAddress.getHostName(), remoteAddress.getPort(),
getLocalAddress().getHostName(), getLocalAddress().getPort()));
- // Start the per-connection read loop on a virtual thread (Java
21+) LAST, so an
- // unchecked throw from the logging/audit above cannot leak an
already-running thread
- // (the catch only handles IOException and does not stop the read
loop).
+ // Start the per-connection read loop on a virtual thread (Java
21+) LAST, so a
+ // throw from the logging/audit above cannot leak an
already-running thread.
this.readThread = Thread.ofVirtual()
.name("TCP-Read-" + remoteAddress.getHostName() + ":" +
remoteAddress.getPort())
.start(this::runReadLoop);
- } catch (IOException e) {
+ } catch (Exception e) {
+ // Close before any other side effect (logging/audit), so a
failure there can't skip cleanup.
+ if (channel != null) {
+ try {
+ channel.close();
+ } catch (IOException closeException) {
+ e.addSuppressed(closeException);
+ }
+ }
String errorMsg = String.format("Failed to connect to %s:%d - %s",
remoteAddress.getHostName(), remoteAddress.getPort(),
e.getMessage());
LOGGER.error(errorMsg, e);
diff --git
a/plc4j/transports/tcp/src/test/java/org/apache/plc4x/java/transport/tcp/TcpTransportInstanceTest.java
b/plc4j/transports/tcp/src/test/java/org/apache/plc4x/java/transport/tcp/TcpTransportInstanceTest.java
index 57c9ea76bd..0781cd7689 100644
---
a/plc4j/transports/tcp/src/test/java/org/apache/plc4x/java/transport/tcp/TcpTransportInstanceTest.java
+++
b/plc4j/transports/tcp/src/test/java/org/apache/plc4x/java/transport/tcp/TcpTransportInstanceTest.java
@@ -290,6 +290,55 @@ class TcpTransportInstanceTest {
assertTrue(doneLatch.await(10, TimeUnit.SECONDS));
}*/
+ @Test
+ void testConstructor_failedBind_doesNotLeakFileDescriptors() throws
Exception {
+ // getOpenFileDescriptorCount() is only exposed on Unix-flavoured JVMs
+ java.lang.management.OperatingSystemMXBean rawOsBean =
+ java.lang.management.ManagementFactory.getOperatingSystemMXBean();
+ org.junit.jupiter.api.Assumptions.assumeTrue(
+ rawOsBean instanceof com.sun.management.UnixOperatingSystemMXBean,
+ "Open file descriptor count not available on this platform");
+ com.sun.management.UnixOperatingSystemMXBean osBean =
+ (com.sun.management.UnixOperatingSystemMXBean) rawOsBean;
+
+ // Occupy a specific local port so binding our own channel to it
deterministically fails
+ // with BindException
+ try (ServerSocketChannel occupier = ServerSocketChannel.open()) {
+ occupier.bind(new InetSocketAddress("127.0.0.1", 0));
+ int occupiedPort = ((InetSocketAddress)
occupier.getLocalAddress()).getPort();
+
+ TcpTransportConfiguration config = new TcpTransportConfiguration();
+ config.receiveBufferSize = 81920;
+ config.localAddress = "127.0.0.1";
+ config.localPort = occupiedPort;
+ // Never actually reached - bind() fails first - but must be a
well-formed address.
+ InetSocketAddress remoteAddress = new
InetSocketAddress("127.0.0.1", occupiedPort);
+
+ // Sample the fd count after every attempt rather than just
before/after
+ int attempts = 20;
+ long previous = osBean.getOpenFileDescriptorCount();
+ int increases = 0;
+ for (int i = 0; i < attempts; i++) {
+ assertThrows(TransportException.class, () ->
+ new TcpTransportInstance(remoteAddress, config,
AuditLog.builder().build())
+ );
+ long current = osBean.getOpenFileDescriptorCount();
+ if (current > previous) {
+ increases++;
+ }
+ previous = current;
+ }
+
+ // Without the fix each failed constructor call leaks the
SocketChannel it opened before
+ // bind() failed, so the fd count grows on essentially every
attempt. With the fix it doesn't.
+ double leakThreshold = 0.75;
+ assertTrue(increases < attempts * leakThreshold,
+ "Open file descriptor count increased on " + increases + "/" +
attempts
+ + " failed bind attempts; expected only incidental growth,
not a consistent"
+ + " per-attempt increase (possible SocketChannel leak)");
+ }
+ }
+
@Test
void testClose_idempotent() throws TransportException {
transportInstance.close();
