This is an automated email from the ASF dual-hosted git repository.
He-Pin pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko.git
The following commit(s) were added to refs/heads/main by this push:
new ceca95fe0d Close unregistered TCP channels on setup failure (#3273)
ceca95fe0d is described below
commit ceca95fe0df3e6a4c5ab035df404053d810f567b
Author: Goutam Adwant <[email protected]>
AuthorDate: Sun Jul 5 11:19:26 2026 -0700
Close unregistered TCP channels on setup failure (#3273)
* Close unregistered TCP channels on setup failure #3246
Motivation:
Outgoing TCP connection setup can fail before the connection actor receives
a ChannelRegistration. In that path the SocketChannel remained open after
CommandFailed was reported.
Modification:
Close still-open channels directly from postStop when no
ChannelRegistration exists. Keep the existing registered-channel cancelAndClose
path unchanged. Extend the setup-failure integration test to verify the socket
is closed.
Result:
Setup failures before ChannelRegistration no longer leave the SocketChannel
open.
Tests:
- sbt "actor-tests / Test / testOnly org.apache.pekko.io.TcpIntegrationSpec"
- sbt "actor / Compile / scalafmtCheck" "actor-tests / Test / scalafmtCheck"
- git diff --check
References:
Fixes #3246
* Address TCP close review comments
---
.../src/test/scala/org/apache/pekko/io/TcpIntegrationSpec.scala | 9 ++++++++-
actor/src/main/scala/org/apache/pekko/io/TcpConnection.scala | 9 ++++++++-
2 files changed, 16 insertions(+), 2 deletions(-)
diff --git
a/actor-tests/src/test/scala/org/apache/pekko/io/TcpIntegrationSpec.scala
b/actor-tests/src/test/scala/org/apache/pekko/io/TcpIntegrationSpec.scala
index e4de5a636e..44e8fe1756 100644
--- a/actor-tests/src/test/scala/org/apache/pekko/io/TcpIntegrationSpec.scala
+++ b/actor-tests/src/test/scala/org/apache/pekko/io/TcpIntegrationSpec.scala
@@ -15,6 +15,7 @@ package org.apache.pekko.io
import java.io.IOException
import java.net.{ InetSocketAddress, ServerSocket, Socket }
+import java.util.concurrent.atomic.AtomicReference
import scala.concurrent.duration._
@@ -187,8 +188,12 @@ class TcpIntegrationSpec extends PekkoSpec("""
"reply with CommandFailed when a connect socket option fails before
connect" in {
val connectCommander = TestProbe()
val failure = new UnsupportedOperationException("boom")
+ val socket = new AtomicReference[Socket]
val failingOption = new Inet.SocketOption {
- override def beforeConnect(s: Socket): Unit = throw failure
+ override def beforeConnect(s: Socket): Unit = {
+ socket.set(s)
+ throw failure
+ }
}
val endpoint = new InetSocketAddress("127.0.0.1", 1)
val command = Connect(endpoint, options = List(failingOption))
@@ -198,6 +203,8 @@ class TcpIntegrationSpec extends PekkoSpec("""
val commandFailed = connectCommander.expectMsgType[CommandFailed]
commandFailed.cmd should ===(command)
commandFailed.cause should ===(Some(failure))
+ socket.get should not be null
+ awaitCond(socket.get.isClosed)
}
"handle tcp connection actor death properly" in new
TestSetup(shouldBindServer = false) {
diff --git a/actor/src/main/scala/org/apache/pekko/io/TcpConnection.scala
b/actor/src/main/scala/org/apache/pekko/io/TcpConnection.scala
index 648ced952e..b866f99ed0 100644
--- a/actor/src/main/scala/org/apache/pekko/io/TcpConnection.scala
+++ b/actor/src/main/scala/org/apache/pekko/io/TcpConnection.scala
@@ -410,8 +410,15 @@ private[io] abstract class TcpConnection(val tcp: TcpExt,
val channel: SocketCha
(if (writePending) Set(pendingWrite.commander) else Set.empty) ++
closedMessage.toList.flatMap(_.notificationsTo).toSet
- if (channel.isOpen) // if channel is still open here, we didn't go through
stopWith => unexpected actor termination
+ if (channel.isOpen) {
prepareAbort()
+ // Without a ChannelRegistration there is no selector callback to close
through.
+ if (registration.isEmpty)
+ try channel.close()
+ catch {
+ case NonFatal(e) => log.error(e, "Error closing SocketChannel")
+ }
+ }
def isCommandFailed: Boolean =
closedMessage.exists(_.closedEvent.isInstanceOf[CommandFailed])
def notifyInterested(): Unit =
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
