This is an automated email from the ASF dual-hosted git repository. He-Pin pushed a commit to branch fix/endpoint-robustness-3163 in repository https://gitbox.apache.org/repos/asf/pekko.git
commit 13efbaf4f557bc72d9d045c066262eea9ce2b7be Author: 虎鸣 <[email protected]> AuthorDate: Wed Jun 24 03:55:12 2026 +0800 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 --- .../main/scala/org/apache/pekko/remote/Endpoint.scala | 16 +++++++++------- .../pekko/remote/TransientSerializationErrorSpec.scala | 7 ++++++- 2 files changed, 15 insertions(+), 8 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 88e68e5270..083a81428d 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,7 +1174,7 @@ 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 [{}]. " + @@ -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 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]
