This is an automated email from the ASF dual-hosted git repository.
pjfanning pushed a commit to branch main
in repository
https://gitbox.apache.org/repos/asf/pekko-persistence-cassandra.git
The following commit(s) were added to refs/heads/main by this push:
new 629928b Add EventsByTagMigration.close to release its tag writers
actor (#494)
629928b is described below
commit 629928b5d970abf9f18dd418e05d37608ae6f221
Author: PJ Fanning <[email protected]>
AuthorDate: Wed Sep 23 08:05:23 2026 +0100
Add EventsByTagMigration.close to release its tag writers actor (#494)
Motivation:
EventsByTagMigration starts a top level tag writers actor in its
constructor and
offers no way to stop it, so the actor and its TagWriter children stay
alive for
the lifetime of the ActorSystem after the migration has finished. The actor
was
also anonymous, which makes it hard to identify in logs.
Modification:
Give the tag writers actor a stable name, in the same style as the one
Reconciliation starts, and add `EventsByTagMigration.close()` to stop it.
Result:
Callers can release the actor once the migration has completed, and the
actor is
recognisable in logs and actor paths.
Tests:
- sbt "core/Test/testOnly
org.apache.pekko.persistence.cassandra.EventsByTagMigrationCloseSpec" - passed
on JDK 17 and JDK 21 (new spec, no Cassandra needed)
- sbt "core/mimaReportBinaryIssues" - passed
- scalafmt --mode diff-ref=upstream/main --list - no changes needed
- sbt "core/Test/testOnly
org.apache.pekko.persistence.cassandra.EventsByTagMigrationSpec" - aborted, no
local Cassandra (docker not running)
References:
None - resource leak found while auditing actor and session lifecycles
---
.../cassandra/EventsByTagMigration.scala | 24 +++++++-
.../cassandra/EventsByTagMigrationCloseSpec.scala | 67 ++++++++++++++++++++++
2 files changed, 90 insertions(+), 1 deletion(-)
diff --git
a/core/src/main/scala/org/apache/pekko/persistence/cassandra/EventsByTagMigration.scala
b/core/src/main/scala/org/apache/pekko/persistence/cassandra/EventsByTagMigration.scala
index 148a373..a486f33 100644
---
a/core/src/main/scala/org/apache/pekko/persistence/cassandra/EventsByTagMigration.scala
+++
b/core/src/main/scala/org/apache/pekko/persistence/cassandra/EventsByTagMigration.scala
@@ -31,6 +31,8 @@ import pekko.stream.scaladsl.{ Sink, Source }
import pekko.util.Timeout
import pekko.{ Done, NotUsed }
import com.datastax.oss.driver.api.core.cql.Row
+
+import java.util.concurrent.atomic.AtomicInteger
import scala.concurrent.ExecutionContext
import scala.concurrent.Future
import scala.concurrent.duration._
@@ -38,6 +40,9 @@ import scala.concurrent.duration._
import pekko.actor.ClassicActorSystemProvider
object EventsByTagMigration {
+
+ private val uniqueActorNameCounter = new AtomicInteger(0)
+
def apply(systemProvider: ClassicActorSystemProvider): EventsByTagMigration =
new EventsByTagMigration(systemProvider)
@@ -78,6 +83,9 @@ object EventsByTagMigration {
}
/**
+ * Each instance starts a tag writers actor that lives until
[[EventsByTagMigration#close]] is called,
+ * so an instance should be closed once the migration it was created for has
completed.
+ *
* @param pluginConfigPath The config namespace where the plugin is
configured, default is `pekko.persistence.cassandra`
*/
class EventsByTagMigration(
@@ -96,7 +104,9 @@ class EventsByTagMigration(
private val taggedPreparedStatements = new
TaggedPreparedStatements(journalStatements, session.prepare)
private val tagWriterSession =
TagWritersSession(session, journalSettings.writeProfile,
journalSettings.readProfile, taggedPreparedStatements)
- private val tagWriters =
system.actorOf(TagWriters.props(eventsByTagSettings.tagWriterSettings,
tagWriterSession))
+ private val tagWriters = system.actorOf(
+ TagWriters.props(eventsByTagSettings.tagWriterSettings, tagWriterSession),
+
s"eventsByTagMigration-tag-writers-${EventsByTagMigration.uniqueActorNameCounter.incrementAndGet()}")
private val tagRecovery =
new CassandraTagRecovery(system, session, settings,
taggedPreparedStatements, tagWriters)
@@ -244,4 +254,16 @@ class EventsByTagMigration(
} yield Done
}
+ /**
+ * Stops the tag writers actor that this instance started. Without this the
actor stays alive for
+ * the lifetime of the `ActorSystem`, so call it once the migration has
completed.
+ *
+ * Any migration still in progress is aborted, so only close after the
futures returned by the
+ * other methods have completed. This instance must not be used again after
it has been closed.
+ *
+ * Calling this more than once has no further effect.
+ */
+ def close(): Unit =
+ system.stop(tagWriters)
+
}
diff --git
a/core/src/test/scala/org/apache/pekko/persistence/cassandra/EventsByTagMigrationCloseSpec.scala
b/core/src/test/scala/org/apache/pekko/persistence/cassandra/EventsByTagMigrationCloseSpec.scala
new file mode 100644
index 0000000..4dc460a
--- /dev/null
+++
b/core/src/test/scala/org/apache/pekko/persistence/cassandra/EventsByTagMigrationCloseSpec.scala
@@ -0,0 +1,67 @@
+/*
+ * 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.persistence.cassandra
+
+import com.typesafe.config.ConfigFactory
+import org.apache.pekko
+import org.scalatest.BeforeAndAfterAll
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpecLike
+import pekko.actor.{ ActorIdentity, ActorSystem, Identify }
+import pekko.testkit.{ ImplicitSender, TestKit }
+
+object EventsByTagMigrationCloseSpec {
+
+ // no Cassandra is needed, the tag writers actor is started eagerly and all
queries are lazy
+ val config = ConfigFactory.parseString("""
+ datastax-java-driver.advanced.reconnect-on-init = false
+ """)
+}
+
+class EventsByTagMigrationCloseSpec
+ extends TestKit(ActorSystem("EventsByTagMigrationCloseSpec",
EventsByTagMigrationCloseSpec.config))
+ with AnyWordSpecLike
+ with Matchers
+ with ImplicitSender
+ with BeforeAndAfterAll {
+
+ override protected def afterAll(): Unit = {
+ shutdown()
+ super.afterAll()
+ }
+
+ "EventsByTagMigration" should {
+
+ "stop its tag writers actor when closed, and tolerate a second close" in {
+ val migration = EventsByTagMigration(system)
+
+ // resolve while this is the only actor in the system matching the
pattern: an actor that
+ // has been stopped can still be in the guardian's children when its
watchers have already
+ // seen Terminated, and the Identify sent to it is answered from dead
letters with an empty
+ // ActorIdentity, which would race with the one from the live actor
+ system.actorSelection("/user/eventsByTagMigration-tag-writers-*") !
Identify(())
+ val tagWriters = watch(expectMsgType[ActorIdentity].ref.get)
+
+ migration.close()
+ expectTerminated(tagWriters)
+
+ // a second close has no further effect
+ migration.close()
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
