This is an automated email from the ASF dual-hosted git repository. He-Pin pushed a commit to branch fix/3255-tcp-halfclose-npe in repository https://gitbox.apache.org/repos/asf/pekko.git
commit 784c1c9856f290543d1da50f857cb9c4106034b3 Author: He-Pin(kerr) <[email protected]> AuthorDate: Sun Jun 28 22:24:13 2026 +0800 fix: avoid NPE in outbound TCP stage when upstream finishes before connect Motivation: TcpConnectionStage.closeConnectionUpstreamFinished() dereferenced the connection ActorRef without a null check on the half-close-disabled path. For an outbound connection whose upstream finishes before the Connected message arrives, connection is still null, so connection ! Close threw a NullPointerException and failed the stage. The connecting handler also always issued ConfirmedClose (half-close) regardless of the halfClose setting once such an early-finished upstream was detected. Modification: - Hoist a connection eq null guard to the top of closeConnectionUpstreamFinished() so neither close branch dereferences a null connection; the close is deferred until the connection is established, mirroring closeConnectionDownstreamFinished(). - Make the connecting handler honor the half-close setting: send Close when half-close is disabled and ConfirmedClose when enabled. Result: Outbound connections with halfClose=false no longer throw a NullPointerException when the upstream finishes before the connection is established, and the early-finish close path respects the configured half-close semantics. Tests: - stream-tests/testOnly org.apache.pekko.stream.io.TcpSpec (28 passed). Added a directional test that fails with the NPE before the fix. References: Fixes #3255 --- .../scala/org/apache/pekko/stream/io/TcpSpec.scala | 25 ++++++++++++++++++++++ .../apache/pekko/stream/impl/io/TcpStages.scala | 20 +++++++++++------ 2 files changed, 39 insertions(+), 6 deletions(-) diff --git a/stream-tests/src/test/scala/org/apache/pekko/stream/io/TcpSpec.scala b/stream-tests/src/test/scala/org/apache/pekko/stream/io/TcpSpec.scala index cd9b3a7367..3d5fb1954b 100644 --- a/stream-tests/src/test/scala/org/apache/pekko/stream/io/TcpSpec.scala +++ b/stream-tests/src/test/scala/org/apache/pekko/stream/io/TcpSpec.scala @@ -519,6 +519,31 @@ class TcpSpec extends StreamSpec(""" binding.unbind() } + "not throw a NullPointerException when full-close is requested and upstream finishes before the connection is established (#3255)" in { + val binding = + Tcp() + .bind("127.0.0.1", 0) + .toMat(Sink.foreach { conn => + conn.flow.join(Flow[ByteString]).run() + })(Keep.left) + .run() + .futureValue + + // `Source.empty` completes synchronously during materialization, i.e. before the outbound + // connection is established (the connection ActorRef is still null at that point). With + // half-close disabled the upstream-finished handler used to dereference that null connection + // (`connection ! Close`) and fail the stage with a NullPointerException. The fix defers the + // close until the connection is established, so the stream must complete normally here. + val result = Source + .empty[ByteString] + .via(Tcp().outgoingConnection(binding.localAddress, halfClose = false)) + .runWith(Sink.ignore) + + result.futureValue should ===(Done) + + binding.unbind() + } + "Echo should work even if server is in full close mode" in { val serverAddress = temporaryServerAddress() diff --git a/stream/src/main/scala/org/apache/pekko/stream/impl/io/TcpStages.scala b/stream/src/main/scala/org/apache/pekko/stream/impl/io/TcpStages.scala index 79c95e9383..3f6f2cac8c 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/impl/io/TcpStages.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/impl/io/TcpStages.scala @@ -322,7 +322,12 @@ private[stream] object ConnectionSourceStage { stageActor.watch(connection) connection ! Register(self, keepOpenOnPeerClosed = true, useResumeWriting = false) if (isAvailable(bytesOut)) connection ! ResumeReading - if (isClosed(bytesIn)) connection ! ConfirmedClose + if (isClosed(bytesIn)) + // Upstream finished before the connection was established. Honor the half-close + // setting: ConfirmedClose only closes the write side (half-close), while Close + // tears down the whole connection. Always sending ConfirmedClose here would keep + // the read side open even when half-close is disabled (issue #3255). + connection ! (if (role.halfClose) ConfirmedClose else Close) else pull(bytesIn) case other => log.warning("Unexpected message to connecting TcpStage: [{}]", other.getClass) } @@ -396,23 +401,26 @@ private[stream] object ConnectionSourceStage { } private def closeConnectionUpstreamFinished(): Unit = { - if (isClosed(bytesOut) || !role.halfClose) { + if (connection eq null) { + // This is an outbound connection for which upstream finished before the connection + // was even established. The close is deferred until the connection is established; + // the `connecting` handler inspects `isClosed(bytesIn)` and closes the connection + // honoring the half-close setting. Dereferencing `connection` here would otherwise + // throw a NullPointerException when half-close is disabled (issue #3255). + } else if (isClosed(bytesOut) || !role.halfClose) { // Reading has stopped before, either because of cancel, or PeerClosed, so just Close now // (or half-close is turned off) if (writeInProgress) connectionClosePending = true // will continue when WriteAck is received and writeBuffer drained else connection ! Close - } else if (connection ne null) { + } else { // We still read, so we only close the write side if (writeInProgress) connectionClosePending = true // will continue when WriteAck is received and writeBuffer drained else connection ! ConfirmedClose } - // Otherwise, this is an outbound connection with half-close enabled for which upstream finished - // before the connection was even established. - // In that case we half-close the connection as soon as it's connected } private def closeConnectionDownstreamFinished(): Unit = { --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
