Replace reflection with MethodHandle access in projection

examples/src/test/java/jdocs/jdbc/JdbcProjectionDocExample.java

@@ -34,9 +34,10 @@
 
 // #jdbc-session-imports
 import org.apache.pekko.projection.jdbc.JdbcSession;
+import java.lang.invoke.MethodHandles;
+import java.sql.Connection;
 import java.sql.DriverManager;
 import java.sql.SQLException;
-import java.sql.Connection;
 
 // #jdbc-session-imports
 
@@ -73,10 +74,11 @@ class PlainJdbcSession implements JdbcSession {
 
     public PlainJdbcSession() {
       try {
-        Class.forName("org.h2.Driver");
+        MethodHandles.Lookup lookup = MethodHandles.lookup();
+        lookup.ensureInitialized(lookup.findClass("org.h2.Driver"));
         this.connection = DriverManager.getConnection("jdbc:h2:mem:test;DB_CLOSE_DELAY=-1");
         connection.setAutoCommit(false);
-      } catch (ClassNotFoundException | SQLException e) {
+      } catch (ReflectiveOperationException | SQLException e) {
         throw new RuntimeException(e);
       }
     }

examples/src/test/scala/docs/jdbc/JdbcProjectionDocExample.scala

@@ -65,7 +65,8 @@ object JdbcProjectionDocExample {
   class PlainJdbcSession extends JdbcSession {
 
     lazy val conn = {
-      Class.forName("org.h2.Driver")
+      val lookup = java.lang.invoke.MethodHandles.lookup()
+      lookup.ensureInitialized(lookup.findClass("org.h2.Driver"))
       val c = DriverManager.getConnection("jdbc:h2:mem:test;DB_CLOSE_DELAY=-1")
       c.setAutoCommit(false)
       c

grpc/src/main/scala/org/apache/pekko/projection/grpc/internal/ConsumerFilterStore.scala

@@ -13,6 +13,7 @@
 
 package org.apache.pekko.projection.grpc.internal
 
+import java.lang.invoke.{ MethodHandle, MethodHandles, MethodType }
 import java.util.ConcurrentModificationException
 import java.util.concurrent.ConcurrentHashMap
 
@@ -54,6 +55,15 @@ import org.slf4j.LoggerFactory
 @InternalApi private[pekko] object ConsumerFilterStore {
   sealed trait Command
 
+  private val lookup = MethodHandles.lookup()
+  private val ddataApplyHandles = new ConcurrentHashMap[Class[?], MethodHandle]
+  private val ddataApplyMethodType =
+    MethodType.methodType(
+      classOf[Behavior[?]],
+      classOf[ConsumerFilterSettings],
+      classOf[String],
+      classOf[ActorRef[ConsumerFilterRegistry.FilterUpdated]])
+
   final case class UpdateFilter(criteria: immutable.Seq[FilterCriteria]) extends Command
 
   final case class GetFilter(replyTo: ActorRef[ConsumerFilter.CurrentFilter]) extends Command
@@ -94,18 +104,25 @@ import org.slf4j.LoggerFactory
       .dynamicAccess
       .getObjectFor[Any]("org.apache.pekko.projection.grpc.internal.DdataConsumerFilterStore") match {
       case Success(companion) =>
-        val applyMethod = companion.getClass.getMethod(
-          "apply",
-          classOf[ConsumerFilterSettings],
-          classOf[String],
-          classOf[ActorRef[ConsumerFilterRegistry.FilterUpdated]])
-        applyMethod.invoke(companion, settings, streamId, notifyUpdatesTo).asInstanceOf[Behavior[Command]]
+        ddataApplyHandle(companion.getClass)
+          .invoke(companion, settings, streamId, notifyUpdatesTo)
+          .asInstanceOf[Behavior[Command]]
       case Failure(exc) =>
         LoggerFactory.getLogger(className).error2("Couldn't create instance of [{}]", className, exc)
         throw exc
     }
   }
 
+  private def ddataApplyHandle(companionClass: Class[?]): MethodHandle = {
+    val cached = ddataApplyHandles.get(companionClass)
+    if (cached ne null) cached
+    else {
+      val handle = lookup.findVirtual(companionClass, "apply", ddataApplyMethodType)
+      val existing = ddataApplyHandles.putIfAbsent(companionClass, handle)
+      if (existing eq null) handle else existing
+    }
+  }
+
 }
 
 /**

grpc/src/main/scala/org/apache/pekko/projection/grpc/internal/ProtoAnySerialization.scala

@@ -13,6 +13,9 @@
 
 package org.apache.pekko.projection.grpc.internal
 
+import java.lang.invoke.{ MethodHandle, MethodHandles, MethodType }
+import java.util.concurrent.ConcurrentHashMap
+
 import scala.collection.concurrent.TrieMap
 import scala.collection.immutable
 import scala.jdk.CollectionConverters._
@@ -43,6 +46,9 @@ import scalapb.options.Scalapb
   final val PekkoSerializationTypeUrlPrefix = "ser.pekko.io/"
   final val PekkoTypeUrlManifestSeparator = ':'
   private final val ProtoAnyTypeUrl = GoogleTypeUrlPrefix + "google.protobuf.Any"
+  private val publicLookup = MethodHandles.publicLookup()
+  private val parserMethodType = MethodType.methodType(classOf[Parser[?]])
+  private val parserHandles = new ConcurrentHashMap[Class[?], MethodHandle]
 
   private val log = LoggerFactory.getLogger(classOf[ProtoAnySerialization])
 
@@ -218,24 +224,21 @@ import scalapb.options.Scalapb
       log.debug("tryResolveJavaPbType attempting to load class {}", className)
 
       val clazz = system.dynamicAccess.getClassFor[Any](className).get
-      val parser = clazz
-        .getMethod("parser")
-        .invoke(null)
-        .asInstanceOf[Parser[com.google.protobuf.Message]]
+      val parser = parserHandle(clazz).invoke().asInstanceOf[Parser[com.google.protobuf.Message]]
       Some(new JavaPbResolvedType(parser))
 
     } catch {
       case cnfe: ClassNotFoundException =>
         log.debug2("Failed to load class [{}] because: {}", className, cnfe.getMessage)
         None
       case nsme: NoSuchElementException =>
-        // Not sure this is exception is thrown. NoSuchMethodException is thrown from getMethod("parser").
+        // Not sure this is exception is thrown. NoSuchMethodException is thrown from the parser MethodHandle lookup.
         // It was like this in the original Kalix JVM SDK.
         throw SerializationException(
           s"Found com.google.protobuf.Message class $className to deserialize protobuf ${typeDescriptor.getFullName} but it didn't have a static parser() method on it.",
           nsme)
       case _: NoSuchMethodException =>
-        // NoSuchMethodException may be thrown from getMethod("parser") if the ScalaPB class can be loaded,
+        // NoSuchMethodException may be thrown from the MethodHandle lookup if the ScalaPB class can be loaded,
         // but the ScalaPB class doesn't have the parser method.
         None
       case iae @ (_: IllegalAccessException | _: IllegalArgumentException) =>
@@ -245,6 +248,16 @@ import scalapb.options.Scalapb
     }
   }
 
+  private def parserHandle(clazz: Class[?]): MethodHandle = {
+    val cached = parserHandles.get(clazz)
+    if (cached ne null) cached
+    else {
+      val handle = publicLookup.findStatic(clazz, "parser", parserMethodType)
+      val existing = parserHandles.putIfAbsent(clazz, handle)
+      if (existing eq null) handle else existing
+    }
+  }
+
   private def hasExtension[ContainerT <: com.google.protobuf.GeneratedMessage.ExtendableMessage[ContainerT], T](
       msg: com.google.protobuf.GeneratedMessage.ExtendableMessageOrBuilder[ContainerT],
       ext: com.google.protobuf.GeneratedMessage.GeneratedExtension[ContainerT, T]

jdbc-int-test/src/test/scala/org/apache/pekko/projection/jdbc/JdbcContainerOffsetStoreSpec.scala

@@ -55,7 +55,8 @@ object JdbcContainerOffsetStoreSpec {
       val container = _container.get
 
       new PureJdbcSession(() => {
-        Class.forName(container.getDriverClassName)
+        val lookup = java.lang.invoke.MethodHandles.lookup()
+        lookup.ensureInitialized(lookup.findClass(container.getDriverClassName))
         val conn =
           DriverManager.getConnection(container.getJdbcUrl, container.getUsername, container.getPassword)
         conn.setAutoCommit(false)

jdbc-int-test/src/test/scala/org/apache/pekko/projection/jdbc/JdbcProjectionSpec.scala

@@ -87,7 +87,8 @@ object JdbcProjectionSpec {
   class PureJdbcSession extends JdbcSession {
 
     lazy val conn = {
-      Class.forName("org.h2.Driver")
+      val lookup = java.lang.invoke.MethodHandles.lookup()
+      lookup.ensureInitialized(lookup.findClass("org.h2.Driver"))
       val c = DriverManager.getConnection("jdbc:h2:mem:jdbc-projection-test;DB_CLOSE_DELAY=-1")
       c.setAutoCommit(false)
       c

jdbc/src/test/java/org/apache/pekko/projection/jdbc/JdbcProjectionTest.java

@@ -18,6 +18,7 @@
 
 import com.typesafe.config.Config;
 import com.typesafe.config.ConfigFactory;
+import java.lang.invoke.MethodHandles;
 import java.sql.Connection;
 import java.sql.DriverManager;
 import java.sql.SQLException;
@@ -87,11 +88,12 @@ static class PureJdbcSession implements JdbcSession {
 
     public PureJdbcSession() {
       try {
-        Class.forName("org.h2.Driver");
+        MethodHandles.Lookup lookup = MethodHandles.lookup();
+        lookup.ensureInitialized(lookup.findClass("org.h2.Driver"));
         Connection c = DriverManager.getConnection("jdbc:h2:mem:test-java;DB_CLOSE_DELAY=-1");
         c.setAutoCommit(false);
         this.connection = c;
-      } catch (ClassNotFoundException | SQLException e) {
+      } catch (ReflectiveOperationException | SQLException e) {
         throw new RuntimeException(e);
       }
     }

jdbc/src/test/scala/org/apache/pekko/projection/jdbc/JdbcOffsetStoreSpec.scala

@@ -101,7 +101,8 @@ object JdbcOffsetStoreSpec {
 
     def jdbcSessionFactory(): PureJdbcSession =
       new PureJdbcSession(() => {
-        Class.forName("org.h2.Driver")
+        val lookup = java.lang.invoke.MethodHandles.lookup()
+        lookup.ensureInitialized(lookup.findClass("org.h2.Driver"))
         val conn = DriverManager.getConnection("jdbc:h2:mem:offset-store-test-jdbc;DB_CLOSE_DELAY=-1")
         conn.setAutoCommit(false)
         conn