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 74b9ab2dec fix: harden EndpointReader against NonFatal dispatch errors
and unwrap WrappedMessage in writer logs (#3166)
74b9ab2dec is described below
commit 74b9ab2dec804085abb119f7dadc67da3251ec8d
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Wed Jun 24 22:02:01 2026 +0800
fix: harden EndpointReader against NonFatal dispatch errors and unwrap
WrappedMessage in writer logs (#3166)
* 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 65eae47858..aab9bef792 100644
--- a/remote/src/main/scala/org/apache/pekko/remote/Endpoint.scala
+++ b/remote/src/main/scala/org/apache/pekko/remote/Endpoint.scala
@@ -921,7 +921,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)")
@@ -948,13 +948,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)
@@ -1162,8 +1162,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 =>
@@ -1183,10 +1182,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 "",
@@ -1246,7 +1245,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 9f0b37ca6a..cf5ff06f2e 100644
---
a/remote/src/test/scala/org/apache/pekko/remote/TransientSerializationErrorSpec.scala
+++
b/remote/src/test/scala/org/apache/pekko/remote/TransientSerializationErrorSpec.scala
@@ -31,6 +31,7 @@ object TransientSerializationErrorSpec {
object ToBinaryIllegal
object NotDeserializable
object IllegalOnDeserialize
+ object RuntimeOnDeserialize
class TestSerializer(@nowarn("msg=never used") system: ExtendedActorSystem)
extends SerializerWithStringManifest {
def identifier: Int = 666
@@ -41,6 +42,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 {
@@ -52,6 +54,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()
}
}
@@ -77,6 +80,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
}
}
}
@@ -110,7 +114,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]
