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 c2aaa25 Add Reconciliation.close to release its tag writers actor
(#492)
c2aaa25 is described below
commit c2aaa25597300dc11c5067bb3986a962d144323b
Author: PJ Fanning <[email protected]>
AuthorDate: Wed Sep 23 08:18:55 2026 +0100
Add Reconciliation.close to release its tag writers actor (#492)
Motivation:
Every Reconciliation instance starts a `reconciliation-tag-writers-N` system
actor and offers no way to stop it, so the actor and its TagWriter children
stay
alive for the lifetime of the ActorSystem. The unique name counter shows
that
repeated instantiation is expected, and the reconciler specs construct one
per
test case, so instances accumulate.
Modification:
Add `Reconciliation.close()`, which stops the tag writers actor, and
document
that an instance should be closed once its operations have completed.
Update the
reconciler doc snippets and the events-by-tag docs to call it.
Result:
Callers can release the actor when they are done reconciling instead of
leaking
it for the lifetime of the ActorSystem.
Tests:
- sbt "core/Test/testOnly
org.apache.pekko.persistence.cassandra.reconciler.ReconciliationCloseSpec" -
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
- Cassandra integration tests not run - no local Cassandra (docker not
running)
References:
None - resource leak found while auditing actor and session lifecycles
---
.../cassandra/reconciler/Reconciliation.scala | 15 +++++
.../AllPersistenceIdsMigrationCompileOnly.scala | 2 +
.../doc/reconciler/ReconciliationCompileOnly.scala | 5 +-
.../reconciler/ReconciliationCloseSpec.scala | 67 ++++++++++++++++++++++
docs/src/main/paradox/events-by-tag.md | 3 +
5 files changed, 91 insertions(+), 1 deletion(-)
diff --git
a/core/src/main/scala/org/apache/pekko/persistence/cassandra/reconciler/Reconciliation.scala
b/core/src/main/scala/org/apache/pekko/persistence/cassandra/reconciler/Reconciliation.scala
index e24b1c3..83ac882 100644
---
a/core/src/main/scala/org/apache/pekko/persistence/cassandra/reconciler/Reconciliation.scala
+++
b/core/src/main/scala/org/apache/pekko/persistence/cassandra/reconciler/Reconciliation.scala
@@ -142,6 +142,9 @@ final private[pekko] class ReconciliationSession(session:
CassandraSession, stat
* To support running in the same system as a journal the tag writers actor
would need to be shared
* and all the interleavings of the actor running at the same be considered.
*
+ * Each instance starts a tag writers actor that lives until
[[Reconciliation#close]] is called, so
+ * an instance should be closed once the reconciliation operations it was
created for have completed.
+ *
* API likely to change when a java/scaladsl is added.
*/
@ApiMayChange
@@ -213,4 +216,16 @@ final class Reconciliation(systemProvider:
ClassicActorSystemProvider, settings:
def rebuildAllPersistenceIds(): Future[Done] =
queries.currentPersistenceIdsFromMessages().runWith(recSession.insertIntoPersistenceIds())
+
+ /**
+ * 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 reconciliation
operations have completed.
+ *
+ * Any operation 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/doc/reconciler/AllPersistenceIdsMigrationCompileOnly.scala
b/core/src/test/scala/doc/reconciler/AllPersistenceIdsMigrationCompileOnly.scala
index 3088c94..34c6a61 100644
---
a/core/src/test/scala/doc/reconciler/AllPersistenceIdsMigrationCompileOnly.scala
+++
b/core/src/test/scala/doc/reconciler/AllPersistenceIdsMigrationCompileOnly.scala
@@ -37,9 +37,11 @@ class AllPersistenceIdsMigrationCompileOnly {
result.onComplete {
case Success(_) =>
system.log.info("All persistenceIds migrated.")
+ rec.close()
system.terminate()
case Failure(e) =>
system.log.error(e, "All persistenceIds migration failed.")
+ rec.close()
system.terminate()
}
// #migrate
diff --git a/core/src/test/scala/doc/reconciler/ReconciliationCompileOnly.scala
b/core/src/test/scala/doc/reconciler/ReconciliationCompileOnly.scala
index 7b49dfe..aed5309 100644
--- a/core/src/test/scala/doc/reconciler/ReconciliationCompileOnly.scala
+++ b/core/src/test/scala/doc/reconciler/ReconciliationCompileOnly.scala
@@ -34,7 +34,7 @@ class ReconciliationCompileOnly {
// Drop and re-create data for a persistence id
val pid = "pid1"
- for {
+ val result: Future[Done] = for {
// do this before dropping the data
tags <- rec.tagsForPersistenceId(pid)
// drop the tag view for every tag for this persistence id
@@ -42,5 +42,8 @@ class ReconciliationCompileOnly {
// optional: re-build, if this is ommited then it will be re-build next
time the pid is started
_ <- rec.rebuildTagViewForPersistenceIds(pid)
} yield Done
+
+ // release the tag writers actor that the Reconciliation started
+ result.onComplete(_ => rec.close())
// #reconcile
}
diff --git
a/core/src/test/scala/org/apache/pekko/persistence/cassandra/reconciler/ReconciliationCloseSpec.scala
b/core/src/test/scala/org/apache/pekko/persistence/cassandra/reconciler/ReconciliationCloseSpec.scala
new file mode 100644
index 0000000..7cf2f87
--- /dev/null
+++
b/core/src/test/scala/org/apache/pekko/persistence/cassandra/reconciler/ReconciliationCloseSpec.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.reconciler
+
+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 ReconciliationCloseSpec {
+
+ // 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 ReconciliationCloseSpec
+ extends TestKit(ActorSystem("ReconciliationCloseSpec",
ReconciliationCloseSpec.config))
+ with AnyWordSpecLike
+ with Matchers
+ with ImplicitSender
+ with BeforeAndAfterAll {
+
+ override protected def afterAll(): Unit = {
+ shutdown()
+ super.afterAll()
+ }
+
+ "Reconciliation" should {
+
+ "stop its tag writers actor when closed, and tolerate a second close" in {
+ val reconciliation = new Reconciliation(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("/system/reconciliation-tag-writers-*") !
Identify(())
+ val tagWriters = watch(expectMsgType[ActorIdentity].ref.get)
+
+ reconciliation.close()
+ expectTerminated(tagWriters)
+
+ // a second close has no further effect
+ reconciliation.close()
+ }
+ }
+}
diff --git a/docs/src/main/paradox/events-by-tag.md
b/docs/src/main/paradox/events-by-tag.md
index e3df245..8c9a966 100644
--- a/docs/src/main/paradox/events-by-tag.md
+++ b/docs/src/main/paradox/events-by-tag.md
@@ -153,6 +153,9 @@ It supports:
After deleting the tag views they will be automatically re-built next time the
persistence id starts or with an explicit rebuild.
+Each `Reconciliation` instance starts a tag writers actor that lives for as
long as the `ActorSystem` does, so call `close()`
+once the reconciliation operations have completed.
+
For example, to rebuild the data for a persistence id:
@@snip
[reconciler](/core/src/test/scala/doc/reconciler/ReconciliationCompileOnly.scala)
{ #imports #reconcile}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
