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 339537eec3 fix: avoid NPE in outbound TCP stage when upstream finishes
before connect (#3259)
339537eec3 is described below
commit 339537eec3b3be39bfba79156e29b2c43b3313b2
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Tue Jul 7 01:37:12 2026 +0800
fix: avoid NPE in outbound TCP stage when upstream finishes before connect
(#3259)
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 75f855b3a2..caef6e4cd4 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]
