This is an automated email from the ASF dual-hosted git repository.
He-Pin pushed a commit to branch 1.7.x
in repository https://gitbox.apache.org/repos/asf/pekko.git
The following commit(s) were added to refs/heads/1.7.x by this push:
new 0f45237159 fix: harden EndpointReader against NonFatal dispatch errors
and unwrap WrappedMessage in writer logs (#3166) (#3169)
0f45237159 is described below
commit 0f45237159b442ac2f2df0795ce49dc749089610
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Thu Jun 25 04:35:00 2026 +0800
fix: harden EndpointReader against NonFatal dispatch errors and unwrap
WrappedMessage in writer logs (#3166) (#3169)
* fix: harden EndpointReader against NonFatal dispatch errors and unwrap
WrappedMessage in writer logs
Motivation:
EndpointReader only caught NotSerializableException and
IllegalArgumentException
during non-reliable message dispatch, allowing other NonFatal exceptions
(e.g.
ClassNotFoundException, RuntimeException from custom serializers) to escape
and
kill the endpoint. The reliable delivery path (deliverAndAck) had no
exception
protection at all. Additionally, EndpointWriter error logs showed the outer
wrapper class instead of the actual message type for wrapped messages.
Modification:
- EndpointReader: widen catch to NonFatal for both non-reliable dispatch and
deliverAndAck paths
- EndpointWriter: use WrappedMessage.unwrap() to log the inner message class
in oversized payload and serialization error messages
- Change logTransientSerializationError parameter from Exception to
Throwable
Result:
All NonFatal deserialization errors are handled as transient (connection
stays
alive) on both reliable and non-reliable delivery paths. Error logs now show
the actual message type for wrapped messages, improving debuggability.
Tests:
- remote / Test / testOnly
org.apache.pekko.remote.TransientSerializationErrorSpec
- Added RuntimeOnDeserialize test case to verify RuntimeException is caught
References:
Fixes #3163
* fix: update log message to cover all deserialization failures
Address review feedback: widen log message from "Serializer not defined"
to "Could not deserialize" since the catch block now handles all
NonFatal exceptions, not just missing serializer configurations.
---
.../main/scala/org/apache/pekko/remote/Endpoint.scala | 18 ++++++++++--------
.../pekko/remote/TransientSerializationErrorSpec.scala | 7 ++++++-
2 files changed, 16 insertions(+), 9 deletions(-)
diff --git a/remote/src/main/scala/org/apache/pekko/remote/Endpoint.scala
b/remote/src/main/scala/org/apache/pekko/remote/Endpoint.scala
index 9da404cdad..785e7ae353 100644
--- a/remote/src/main/scala/org/apache/pekko/remote/Endpoint.scala
+++ b/remote/src/main/scala/org/apache/pekko/remote/Endpoint.scala
@@ -913,7 +913,7 @@ private[remote] class EndpointWriter(
if (pduSize > transport.maximumPayloadBytes) {
val reasonText =
s"Discarding oversized payload sent to ${s.recipient}: max
allowed size ${transport
- .maximumPayloadBytes} bytes, actual size of encoded
${s.message.getClass} was ${pdu.size} bytes."
+ .maximumPayloadBytes} bytes, actual size of encoded
${WrappedMessage.unwrap(s.message).getClass} was ${pdu.size} bytes."
log.error(
new OversizedPayloadException(reasonText),
"Transient association error (association remains live)")
@@ -940,13 +940,13 @@ private[remote] class EndpointWriter(
log.error(
e,
"Serializer not defined for message type [{}]. Transient association
error (association remains live)",
- s.message.getClass)
+ WrappedMessage.unwrap(s.message).getClass)
true
case e: IllegalArgumentException =>
log.error(
e,
"Serializer not defined for message type [{}]. Transient association
error (association remains live)",
- s.message.getClass)
+ WrappedMessage.unwrap(s.message).getClass)
true
case e: MessageSerializer.SerializationException =>
log.error(e, "{} Transient association error (association remains
live)", e.getMessage)
@@ -1154,8 +1154,7 @@ private[remote] class EndpointReader(
} else
try msgDispatch.dispatch(msg.recipient, msg.recipientAddress,
msg.serializedMessage, msg.senderOption)
catch {
- case e: NotSerializableException =>
logTransientSerializationError(msg, e)
- case e: IllegalArgumentException =>
logTransientSerializationError(msg, e)
+ case NonFatal(e) => logTransientSerializationError(msg, e)
}
case None =>
@@ -1175,10 +1174,10 @@ private[remote] class EndpointReader(
}
- private def logTransientSerializationError(msg: PekkoPduCodec.Message,
error: Exception): Unit = {
+ private def logTransientSerializationError(msg: PekkoPduCodec.Message,
error: Throwable): Unit = {
val sm = msg.serializedMessage
log.warning(
- "Serializer not defined for message with serializer id [{}] and manifest
[{}]. " +
+ "Could not deserialize message with serializer id [{}] and manifest
[{}]. " +
"Transient association error (association remains live). {}",
sm.getSerializerId,
if (sm.hasMessageManifest) sm.getMessageManifest.toStringUtf8 else "",
@@ -1238,7 +1237,10 @@ private[remote] class EndpointReader(
// Notify writer that some messages can be acked
context.parent ! OutboundAck(ack)
deliver.foreach { m =>
- msgDispatch.dispatch(m.recipient, m.recipientAddress,
m.serializedMessage, m.senderOption)
+ try msgDispatch.dispatch(m.recipient, m.recipientAddress,
m.serializedMessage, m.senderOption)
+ catch {
+ case NonFatal(e) => logTransientSerializationError(m, e)
+ }
}
}
diff --git
a/remote/src/test/scala/org/apache/pekko/remote/TransientSerializationErrorSpec.scala
b/remote/src/test/scala/org/apache/pekko/remote/TransientSerializationErrorSpec.scala
index 001906e21f..01879af2d1 100644
---
a/remote/src/test/scala/org/apache/pekko/remote/TransientSerializationErrorSpec.scala
+++
b/remote/src/test/scala/org/apache/pekko/remote/TransientSerializationErrorSpec.scala
@@ -30,6 +30,7 @@ object TransientSerializationErrorSpec {
object ToBinaryIllegal
object NotDeserializable
object IllegalOnDeserialize
+ object RuntimeOnDeserialize
class TestSerializer(@unused system: ExtendedActorSystem) extends
SerializerWithStringManifest {
def identifier: Int = 666
@@ -40,6 +41,7 @@ object TransientSerializationErrorSpec {
case ToBinaryIllegal => "TI"
case NotDeserializable => "ND"
case IllegalOnDeserialize => "IOD"
+ case RuntimeOnDeserialize => "ROD"
case _ => throw new NotSerializableException()
}
def toBinary(o: AnyRef): Array[Byte] = o match {
@@ -51,6 +53,7 @@ object TransientSerializationErrorSpec {
manifest match {
case "ND" => throw new NotSerializableException() // Not sure this
applies here
case "IOD" => throw new IllegalArgumentException()
+ case "ROD" => throw new RuntimeException("simulated deserialization
failure")
case _ => throw new NotSerializableException()
}
}
@@ -76,6 +79,7 @@ abstract class
AbstractTransientSerializationErrorSpec(config: Config)
"org.apache.pekko.remote.TransientSerializationErrorSpec$ToBinaryIllegal$" =
test
"org.apache.pekko.remote.TransientSerializationErrorSpec$NotDeserializable$" =
test
"org.apache.pekko.remote.TransientSerializationErrorSpec$IllegalOnDeserialize$"
= test
+
"org.apache.pekko.remote.TransientSerializationErrorSpec$RuntimeOnDeserialize$"
= test
}
}
}
@@ -109,7 +113,8 @@ abstract class
AbstractTransientSerializationErrorSpec(config: Config)
ToBinaryIllegal,
ToBinaryNotSerializable,
NotDeserializable,
- IllegalOnDeserialize).foreach(msg => selection.tell(msg,
this.testActor))
+ IllegalOnDeserialize,
+ RuntimeOnDeserialize).foreach(msg => selection.tell(msg,
this.testActor))
// make sure we still have a connection
selection.tell("ping", this.testActor)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
