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 dce94d9a0b Fix ScalaPB generated-message deserialization #3665 (#3665)
dce94d9a0b is described below
commit dce94d9a0b7612a72944640a8a5deac19a69045a
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Fri Oct 23 23:17:08 2026 +0800
Fix ScalaPB generated-message deserialization #3665 (#3665)
Motivation:
The default ScalaPB binding can encode generated messages, but decoding
fails when `parseFrom` returns the erased `GeneratedMessage` type.
Modification:
Resolve the parser's declared return type, require compatibility with the
manifest, and check the actual returned instance while preserving
target-exception wrapping. Cover generated-message round trips, invalid parser
results, binding overrides, and startup without ScalaPB; the dependency is
test-only.
Result:
ScalaPB messages round-trip through the protobuf serializer without
weakening its manifest type contract or requiring ScalaPB at runtime.
Tests:
Validation used a local integration checkout with current main and the
other pending PRs; MiMa ran on this rebased branch.
- `remote / Test / testOnly
org.apache.pekko.remote.serialization.ProtobufSerializerSpec
org.apache.pekko.remote.serialization.ScalaPbSerializationSpec`: 17 passed on
JDK 25; both new parser regressions failed before the fix.
- `+remote/mimaReportBinaryIssues`, `headerCreateAll`, `+headerCheckAll`,
native scalafmt, and `git diff --check`: passed.
- Full validation is delegated to remote CI.
References:
Refs #3651 and
[akka/akka-core#32070](https://github.com/akka/akka-core/pull/32070).
---
project/Dependencies.scala | 4 +-
.../remote/serialization/ProtobufSerializer.scala | 15 ++-
.../serialization/ProtobufSerializerSpec.scala | 37 +++++--
.../serialization/ScalaPbSerializationSpec.scala | 121 +++++++++++++++++++++
4 files changed, 166 insertions(+), 11 deletions(-)
diff --git a/project/Dependencies.scala b/project/Dependencies.scala
index 533da66d57..a4d46c3159 100644
--- a/project/Dependencies.scala
+++ b/project/Dependencies.scala
@@ -184,6 +184,7 @@ object Dependencies {
.exclude("org.testng", "testng")
val protobufRuntime = "com.google.protobuf" % "protobuf-java" %
Protobuf.protobufJavaVersion % Test
+ val scalaPbRuntime = "com.thesamet.scalapb" %% "scalapb-runtime" %
"0.11.21" % Test
// YCSB (Yahoo Cloud Serving Benchmark https://ycsb.site)
val ycsb = "site.ycsb" % "core" % "0.17.0" % Test
@@ -265,7 +266,8 @@ object Dependencies {
TestDependencies.junit,
TestDependencies.scalatest,
TestDependencies.jimfs,
- TestDependencies.protobufRuntime) ++ remoteOptionalDependencies
+ TestDependencies.protobufRuntime,
+ TestDependencies.scalaPbRuntime) ++ remoteOptionalDependencies
lazy val remoteTests = l ++= Seq(TestDependencies.junit,
TestDependencies.scalatest) ++ remoteDependencies
diff --git
a/remote/src/main/scala/org/apache/pekko/remote/serialization/ProtobufSerializer.scala
b/remote/src/main/scala/org/apache/pekko/remote/serialization/ProtobufSerializer.scala
index c45442cbec..d29025b56d 100644
---
a/remote/src/main/scala/org/apache/pekko/remote/serialization/ProtobufSerializer.scala
+++
b/remote/src/main/scala/org/apache/pekko/remote/serialization/ProtobufSerializer.scala
@@ -39,13 +39,24 @@ object ProtobufSerializer {
MethodType.methodType(Void.TYPE, classOf[Throwable]))
private def parsingHandle(clazz: Class[?]): MethodHandle = {
- val methodType = MethodType.methodType(clazz, classOf[Array[Byte]])
+ // ScalaPB's static forwarder can return GeneratedMessage rather than the
concrete message class.
+ val method =
+ try clazz.getDeclaredMethod("parseFrom", classOf[Array[Byte]])
+ catch {
+ case _: NoSuchMethodException => clazz.getMethod("parseFrom",
classOf[Array[Byte]])
+ }
+ val returnType = method.getReturnType
+ require(
+ returnType.isAssignableFrom(clazz),
+ s"parseFrom of [${clazz.getName}] returns unrelated type
[${returnType.getName}]")
+ val methodType = MethodType.methodType(returnType, classOf[Array[Byte]])
val handle =
try lookup.findStatic(clazz, "parseFrom", methodType)
catch {
case _: IllegalAccessException => publicLookup.findStatic(clazz,
"parseFrom", methodType)
}
- val invoker = wrapTargetException(handle).asType(parsingInvokerType)
+ val typedHandle = handle.asType(MethodType.methodType(clazz,
classOf[Array[Byte]]))
+ val invoker = wrapTargetException(typedHandle).asType(parsingInvokerType)
ensureInitialized(clazz)
invoker
}
diff --git
a/remote/src/test/scala/org/apache/pekko/remote/serialization/ProtobufSerializerSpec.scala
b/remote/src/test/scala/org/apache/pekko/remote/serialization/ProtobufSerializerSpec.scala
index 29cc72159f..7921a8d036 100644
---
a/remote/src/test/scala/org/apache/pekko/remote/serialization/ProtobufSerializerSpec.scala
+++
b/remote/src/test/scala/org/apache/pekko/remote/serialization/ProtobufSerializerSpec.scala
@@ -39,6 +39,17 @@ class ThrowingToByteArrayMessage {
def toByteArray(): Array[Byte] = throw new
IllegalArgumentException("serialize-user-bug")
}
+object UnrelatedParseMessage {
+ def parseFrom(@nowarn("msg=never used") bytes: Array[Byte]): String =
"unrelated"
+}
+case class UnrelatedParseMessage() {}
+
+object WrongParseResultMessage {
+ def parseFrom(@nowarn("msg=never used") bytes: Array[Byte]):
ProtobufSerializerSpec.AnotherBase =
+ new AnotherMessage3
+}
+case class WrongParseResultMessage() extends
ProtobufSerializerSpec.AnotherBase {}
+
object ProtobufSerializerSpec {
trait AnotherInterface
abstract class AnotherBase
@@ -76,6 +87,7 @@ class ProtobufSerializerSpec extends PekkoSpec(s"""
"${classOf[AnotherMessage].getName}",
"${classOf[FailingInitializerMessage].getName}",
"${classOf[ThrowingParseMessage].getName}",
+ "${classOf[UnrelatedParseMessage].getName}",
"${classOf[ProtobufSerializerSpec.AnotherInterface].getName}",
"${classOf[ProtobufSerializerSpec.AnotherBase].getName}"
]
@@ -91,14 +103,6 @@ class ProtobufSerializerSpec extends PekkoSpec(s"""
ser.serializerFor(classOf[MyMessageV3]).getClass should
===(classOf[ProtobufSerializer])
}
- "skip the optional ScalaPB binding when ScalaPB is not on the classpath"
in {
- val scalaPbClassName = "scalapb.GeneratedMessage"
- ser.settings.SerializationBindings.get(scalaPbClassName) should
===(Some("proto"))
- val dynamicAccess =
system.asInstanceOf[ExtendedActorSystem].dynamicAccess
- dynamicAccess.getClassFor[Any](scalaPbClassName).isFailure should
===(true)
- ser.bindings.exists { case (clazz, _) => clazz.getName ==
scalaPbClassName } should ===(false)
- }
-
"work for SerializedMessage (just an org.apache.pekko.protobuf message)"
in {
// create a protobuf message
val protobufMessage: SerializedMessage =
@@ -173,6 +177,23 @@ class ProtobufSerializerSpec extends PekkoSpec(s"""
deserialized.getClass should ===(classOf[AnotherMessage3])
}
+ "reject a parser whose declared return type is unrelated to the manifest"
in {
+ val serializer =
ser.serializerFor(classOf[MyMessage]).asInstanceOf[ProtobufSerializer]
+
+ intercept[IllegalArgumentException] {
+ serializer.fromBinary(Array.emptyByteArray,
Some(classOf[UnrelatedParseMessage]))
+ }.getMessage should include("returns unrelated type")
+ }
+
+ "reject a parser result that does not match the manifest and wrap the
target failure" in {
+ val serializer =
ser.serializerFor(classOf[MyMessage]).asInstanceOf[ProtobufSerializer]
+
+ val exception = intercept[InvocationTargetException] {
+ serializer.fromBinary(Array.emptyByteArray,
Some(classOf[WrongParseResultMessage]))
+ }
+ exception.getCause shouldBe a[ClassCastException]
+ }
+
"preserve reflection-compatible target exception wrapping" in {
val serializer =
ser.serializerFor(classOf[MyMessage]).asInstanceOf[ProtobufSerializer]
diff --git
a/remote/src/test/scala/org/apache/pekko/remote/serialization/ScalaPbSerializationSpec.scala
b/remote/src/test/scala/org/apache/pekko/remote/serialization/ScalaPbSerializationSpec.scala
new file mode 100644
index 0000000000..c14c885111
--- /dev/null
+++
b/remote/src/test/scala/org/apache/pekko/remote/serialization/ScalaPbSerializationSpec.scala
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.remote.serialization
+
+import java.io.File
+import java.nio.file.Files
+import java.util.concurrent.TimeUnit
+
+import scala.concurrent.Await
+import scala.concurrent.duration._
+
+import com.google.protobuf.timestamp.Timestamp
+
+import org.apache.pekko
+import pekko.actor.{ ActorSystem, ExtendedActorSystem }
+import pekko.serialization.{ SerializationExtension, Serializers }
+import pekko.testkit.PekkoSpec
+
+import com.typesafe.config.ConfigFactory
+
+class ScalaPbOverrideSerializer(system: ExtendedActorSystem) extends
ProtobufSerializer(system)
+
+object ScalaPbAbsentClasspath {
+ def main(args: Array[String]): Unit = {
+ require(args.isEmpty)
+
require(scala.util.Try(Class.forName("scalapb.GeneratedMessage")).isFailure,
"ScalaPB must be absent")
+ val system = ActorSystem("without-scalapb")
+ try {
+ val serialization = SerializationExtension(system)
+ require(serialization.deserialize(serialization.serialize("still
works").get, classOf[String]).get ==
+ "still works")
+ println("ActorSystem and serialization started without ScalaPB")
+ } finally Await.result(system.terminate(), 15.seconds)
+ }
+}
+
+class ScalaPbSerializationSpec extends PekkoSpec {
+ private val serialization = SerializationExtension(system)
+ private val message = Timestamp(seconds = 123456789L, nanos = 123456789)
+
+ "Optional ScalaPB serialization" must {
+ "select the default binding and round-trip an actual generated message" in
{
+ serialization.findSerializerFor(message).getClass shouldBe
classOf[ProtobufSerializer]
+ val bytes = serialization.serialize(message).get
+ serialization.deserialize(bytes, classOf[Timestamp]).get shouldBe message
+ val serializer = serialization.findSerializerFor(message)
+ serialization.deserialize(bytes, serializer.identifier,
+ Serializers.manifestFor(serializer, message)).get shouldBe message
+ }
+
+ "honor an application override of the marker binding" in {
+ val config = ConfigFactory.parseString(s"""
+ pekko.actor.serializers.scalapb-custom =
"${classOf[ScalaPbOverrideSerializer].getName}"
+
pekko.actor.serialization-identifiers."${classOf[ScalaPbOverrideSerializer].getName}"
= 987654
+ pekko.actor.serialization-bindings."scalapb.GeneratedMessage" =
scalapb-custom
+ """).withFallback(system.settings.config)
+ val other = ActorSystem("scalapb-override", config)
+ try {
+ val overridden = SerializationExtension(other)
+ overridden.findSerializerFor(message).getClass shouldBe
classOf[ScalaPbOverrideSerializer]
+ overridden.deserialize(overridden.serialize(message).get,
classOf[Timestamp]).get shouldBe message
+ } finally shutdown(other)
+ }
+
+ "honor disabling the marker binding" in {
+ val config = ConfigFactory.parseString("""
+ pekko.actor.serialization-bindings."scalapb.GeneratedMessage" = none
+ """).withFallback(system.settings.config)
+ val other = ActorSystem("scalapb-disabled", config)
+ try {
+ val disabled = SerializationExtension(other)
+ disabled.findSerializerFor(message) should not be a[ProtobufSerializer]
+ disabled.serialize(message).isFailure shouldBe true
+ } finally shutdown(other)
+ }
+
+ "start with a real classpath that does not contain the optional runtime"
in {
+ val classpath =
System.getProperty("java.class.path").split(File.pathSeparator).toVector
+ val runtime = new
File(classOf[scalapb.GeneratedMessage].getProtectionDomain.getCodeSource.getLocation.toURI)
+ classpath.map(new File(_).getCanonicalFile) should
contain(runtime.getCanonicalFile)
+ val withoutRuntime = classpath.filterNot(path => new
File(path).getCanonicalFile == runtime.getCanonicalFile)
+ val output = Files.createTempFile("pekko-without-scalapb-", ".log")
+ val java = new File(new File(System.getProperty("java.home"), "bin"),
"java").getPath
+ val process = new ProcessBuilder(
+ java,
+ "-Xmx256m",
+ "-cp",
+ withoutRuntime.mkString(File.pathSeparator),
+ ScalaPbAbsentClasspath.getClass.getName.stripSuffix("$"))
+ .redirectErrorStream(true)
+ .redirectOutput(output.toFile)
+ .start()
+ try {
+ val exited = process.waitFor(30, TimeUnit.SECONDS)
+ withClue(Files.readString(output)) {
+ exited shouldBe true
+ process.exitValue() shouldBe 0
+ Files.readString(output) should include("ActorSystem and
serialization started without ScalaPB")
+ }
+ } finally {
+ if (process.isAlive) process.destroyForcibly().waitFor(10,
TimeUnit.SECONDS)
+ Files.deleteIfExists(output)
+ }
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
