Merge pull request #28598 from akka/wip-PluginHolder-patriknw

Don't start journal when accessing event adapters, #28597

akka-persistence/src/main/mima-filters/2.6.3.backwards.excludes/28597-lazy-journal.excludes

@@ -0,0 +1,2 @@
+# #28597 Don't start journal when accessing event adapters
+ProblemFilters.exclude[Problem]("akka.persistence.Persistence#PluginHolder*")

akka-persistence/src/main/scala/akka/persistence/Persistence.scala

@@ -151,8 +151,11 @@ object Persistence extends ExtensionId[Persistence] with ExtensionIdProvider {
   def lookup() = Persistence
 
   /** INTERNAL API. */
-  private[persistence] case class PluginHolder(actor: ActorRef, adapters: EventAdapters, config: Config)
+  private[persistence] case class PluginHolder(actorFactory: () => ActorRef, adapters: EventAdapters, config: Config)
-      extends Extension
+      extends Extension {
+    // lazy creation of actor so that it's not started when only looking up adapters
+    lazy val actor: ActorRef = actorFactory()
+  }
 
   /** Config path to fall-back to if a setting is not defined in a specific plugin's config section */
   val JournalFallbackConfigPath = "akka.persistence.journal-plugin-fallback"
@@ -428,10 +431,10 @@ class Persistence(val system: ExtendedActorSystem) extends Extension {
         !isEmpty(configPath) && mergedConfig.hasPath(configPath),
         s"'reference.conf' is missing persistence plugin config path: '$configPath'")
       val config: Config = mergedConfig.getConfig(configPath).withFallback(mergedConfig.getConfig(fallbackPath))
-      val plugin: ActorRef = createPlugin(configPath, config)
+      val pluginActorFactory = () => createPlugin(configPath, config)
       val adapters: EventAdapters = createAdapters(configPath, mergedConfig)
 
-      PluginHolder(plugin, adapters, config)
+      PluginHolder(pluginActorFactory, adapters, config)
     }
   }
 

akka-persistence/src/test/scala/akka/persistence/EndToEndEventAdapterSpec.scala

@@ -8,7 +8,7 @@ import java.io.File
 
 import akka.actor._
 import akka.persistence.journal.{ EventAdapter, EventSeq }
-import akka.testkit.{ EventFilter, TestProbe }
+import akka.testkit.TestProbe
 import akka.util.unused
 import com.typesafe.config.{ Config, ConfigFactory }
 import org.apache.commons.io.FileUtils
@@ -233,11 +233,9 @@ abstract class EndToEndEventAdapterSpec(journalName: String, journalConfig: Conf
           s"""$journalPath.event-adapter-bindings."${classOf[EndToEndEventAdapterSpec].getCanonicalName}$$A"""")
 
       withActorSystem("MissingAdapterSystem", journalConfig.withFallback(missingAdapterConfig)) { implicit system2 =>
-        EventFilter[ActorInitializationException](occurrences = 1, pattern = ".*undefined event-adapter.*").intercept {
+        intercept[IllegalArgumentException] {
-          intercept[IllegalArgumentException] {
+          Persistence(system2).adaptersFor(s"akka.persistence.journal.$journalName").get(classOf[String])
-            Persistence(system2).adaptersFor(s"akka.persistence.journal.$journalName").get(classOf[String])
+        }.getMessage should include("was bound to undefined event-adapter: a (bindings: [a, b], known adapters: b)")
-          }.getMessage should include("was bound to undefined event-adapter: a (bindings: [a, b], known adapters: b)")
-        }
       }
     }
   }

akka-persistence/src/test/scala/akka/persistence/LoadPluginSpec.scala

@@ -8,8 +8,9 @@ import akka.persistence.journal.inmem.InmemJournal
 import com.typesafe.config.Config
 import akka.testkit.ImplicitSender
 import akka.actor.Actor
+import akka.util.unused
 
-object LoadJournalSpec {
+object LoadPluginSpec {
 
   case object GetConfig
 
@@ -18,25 +19,44 @@ object LoadJournalSpec {
       case GetConfig => sender() ! config
     }
   }
+
+  object JournalWithStartupNotification {
+    final case class Started(configPath: String)
+  }
+  class JournalWithStartupNotification(@unused config: Config, configPath: String) extends InmemJournal {
+    context.system.eventStream.publish(JournalWithStartupNotification.Started(configPath))
+  }
 }
 
-class LoadJournalSpec
+class LoadPluginSpec
     extends PersistenceSpec(
       PersistenceSpec.config(
         "inmem",
         "LoadJournalSpec",
         extraConfig = Some("""
-  akka.persistence.journal.inmem.class = "akka.persistence.LoadJournalSpec$JournalWithConfig"
+  akka.persistence.journal.inmem.class = "akka.persistence.LoadPluginSpec$JournalWithConfig"
   akka.persistence.journal.inmem.extra-property = 17
+  
+  test-plugin {
+    class = "akka.persistence.LoadPluginSpec$JournalWithStartupNotification"
+  }
   """)))
     with ImplicitSender {
-  import LoadJournalSpec._
+  import LoadPluginSpec._
 
-  "A journal with config parameter" must {
+  "A journal" must {
     "be created with plugin config" in {
       val journalRef = Persistence(system).journalFor("akka.persistence.journal.inmem")
       journalRef ! GetConfig
       expectMsgType[Config].getInt("extra-property") should be(17)
     }
+
+    "not be created via eventAdapter" in {
+      system.eventStream.subscribe(testActor, classOf[JournalWithStartupNotification.Started])
+      Persistence(system).adaptersFor("test-plugin")
+      expectNoMessage()
+      Persistence(system).journalFor("test-plugin")
+      expectMsg(JournalWithStartupNotification.Started("test-plugin"))
+    }
   }
 }