This is an automated email from the ASF dual-hosted git repository. He-Pin pushed a commit to branch backport/3166-endpoint-reader-1.7 in repository https://gitbox.apache.org/repos/asf/pekko.git
commit b384b009a1bfbeb48b275451bd74cb82238ed0d2 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 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]
