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 7342e07f Add Cleanup.close to release the reconciliation it starts
(#493)
7342e07f is described below
commit 7342e07f67e23f7533c09b13bb797c385e9cfbbc
Author: PJ Fanning <[email protected]>
AuthorDate: Wed Sep 23 09:23:48 2026 +0100
Add Cleanup.close to release the reconciliation it starts (#493)
Motivation:
Cleanup's tagged event operations lazily create a Reconciliation, which
starts a
system actor that stays alive for the lifetime of the ActorSystem. Cleanup
is
documented as a per-batch tool that can be run several times in parallel, so
every instance that deletes tagged events leaks one of those actors.
Modification:
Add `Cleanup.close()`, which closes the internal Reconciliation. Track
whether
the lazily created Reconciliation exists so that closing a Cleanup that
never
touched tagged events does not create one just to stop it again. Document
the
new method in the cleanup docs.
Result:
Callers can release the actor when they are done cleaning up, and a Cleanup
used
only for snapshot or event deletes never starts one at all.
Tests:
- sbt "core/Test/testOnly
org.apache.pekko.persistence.cassandra.cleanup.CleanupCloseSpec" - passed (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:
Refs the Reconciliation.close change this builds on - resource leak found
while
auditing actor and session lifecycles
---
.../persistence/cassandra/cleanup/Cleanup.scala | 30 +++++++-
.../cassandra/cleanup/CleanupCloseSpec.scala | 83 ++++++++++++++++++++++
docs/src/main/paradox/cleanup.md | 3 +
3 files changed, 113 insertions(+), 3 deletions(-)
diff --git
a/core/src/main/scala/org/apache/pekko/persistence/cassandra/cleanup/Cleanup.scala
b/core/src/main/scala/org/apache/pekko/persistence/cassandra/cleanup/Cleanup.scala
index d9a0c0e1..edb5a348 100644
---
a/core/src/main/scala/org/apache/pekko/persistence/cassandra/cleanup/Cleanup.scala
+++
b/core/src/main/scala/org/apache/pekko/persistence/cassandra/cleanup/Cleanup.scala
@@ -48,6 +48,9 @@ import com.datastax.oss.driver.api.core.cql.Row
* When a list of `persistenceIds` are given they are deleted sequentially in
the order
* of the list. It's possible to parallelize the deletes by running several
cleanup operations
* at the same time operating on different sets of `persistenceIds`.
+ *
+ * The tagged event operations start an actor that lives until
[[Cleanup#close]] is called, so an
+ * instance should be closed once the cleanup operations it was created for
have completed.
*/
@ApiMayChange
final class Cleanup(systemProvider: ClassicActorSystemProvider, settings:
CleanupSettings) {
@@ -67,9 +70,16 @@ final class Cleanup(systemProvider:
ClassicActorSystemProvider, settings: Cleanu
// operations on journal, snapshotStore and tagViews should be only be done
when dry-run = false
private val journal: ActorRef =
Persistence(system).journalFor(pluginLocation + ".journal")
private lazy val snapshotStore: ActorRef =
Persistence(system).snapshotStoreFor(pluginLocation + ".snapshot")
- private lazy val tagViewsReconciliation = new Reconciliation(
- system,
- new ReconciliationSettings(system.settings.config.getConfig(pluginLocation
+ ".reconciler")))
+ // only the tagged event operations need this and it starts an actor, so
track whether it was
+ // created to keep `close` from creating one just to stop it again
+ @volatile private var tagViewsReconciliationCreated = false
+ private lazy val tagViewsReconciliation = {
+ val reconciliation = new Reconciliation(
+ system,
+ new
ReconciliationSettings(system.settings.config.getConfig(pluginLocation +
".reconciler")))
+ tagViewsReconciliationCreated = true
+ reconciliation
+ }
private implicit val askTimeout: Timeout = operationTimeout
@@ -339,6 +349,20 @@ final class Cleanup(systemProvider:
ClassicActorSystemProvider, settings: Cleanu
sendToSnapshotStore(CassandraSnapshotStore.DeleteAllSnapshots(persistenceId))
}
+ /**
+ * Releases the resources that the tagged event operations acquired. Without
this the actor that
+ * they start stays alive for the lifetime of the `ActorSystem`, so call
this once the cleanup
+ * 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, and it does nothing if
no tagged event
+ * operation was used.
+ */
+ def close(): Unit =
+ if (tagViewsReconciliationCreated) tagViewsReconciliation.close()
+
private def foreach(
persistenceIds: Seq[String],
operationName: String,
diff --git
a/core/src/test/scala/org/apache/pekko/persistence/cassandra/cleanup/CleanupCloseSpec.scala
b/core/src/test/scala/org/apache/pekko/persistence/cassandra/cleanup/CleanupCloseSpec.scala
new file mode 100644
index 00000000..69f18b5d
--- /dev/null
+++
b/core/src/test/scala/org/apache/pekko/persistence/cassandra/cleanup/CleanupCloseSpec.scala
@@ -0,0 +1,83 @@
+/*
+ * 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.cleanup
+
+import scala.concurrent.duration._
+
+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, ActorRef, ActorSystem, Identify }
+import pekko.testkit.{ ImplicitSender, TestKit }
+
+object CleanupCloseSpec {
+
+ // no Cassandra is needed, only the lifecycle of the reconciliation actor is
asserted on
+ val config = ConfigFactory.parseString("""
+ pekko.persistence.cassandra.cleanup.dry-run = true
+ datastax-java-driver.advanced.reconnect-on-init = false
+ """)
+}
+
+class CleanupCloseSpec
+ extends TestKit(ActorSystem("CleanupCloseSpec", CleanupCloseSpec.config))
+ with AnyWordSpecLike
+ with Matchers
+ with ImplicitSender
+ with BeforeAndAfterAll {
+
+ override protected def afterAll(): Unit = {
+ shutdown()
+ super.afterAll()
+ }
+
+ private def reconciliationTagWriters(): Seq[ActorRef] = {
+ system.actorSelection("/system/reconciliation-tag-writers-*") !
Identify(())
+ receiveWhile(500.millis) {
+ case ActorIdentity(_, ref) => ref
+ }.flatten
+ }
+
+ "Cleanup" should {
+
+ "not start a reconciliation when only snapshot and event operations are
used" in {
+ val cleanup = new Cleanup(system)
+
+ cleanup.close()
+
+ reconciliationTagWriters() shouldBe empty
+ }
+
+ "stop the reconciliation it started when closed" in {
+ val cleanup = new Cleanup(system)
+ // the returned future needs Cassandra and is deliberately not awaited,
it is only used to
+ // trigger the creation of the internal Reconciliation
+ cleanup.deleteAllTaggedEvents("pid-1")
+
+ val tagWriters = reconciliationTagWriters()
+ tagWriters should have size 1
+ watch(tagWriters.head)
+
+ cleanup.close()
+
+ expectTerminated(tagWriters.head)
+ }
+ }
+}
diff --git a/docs/src/main/paradox/cleanup.md b/docs/src/main/paradox/cleanup.md
index 2ea58865..f90fa01d 100644
--- a/docs/src/main/paradox/cleanup.md
+++ b/docs/src/main/paradox/cleanup.md
@@ -28,4 +28,7 @@ Scala
Java
: @@snip
[snapshot-keyspace](/docs/src/test/java/jdoc/cleanup/CleanupDocExample.java) {
#cleanup }
+The operations that delete tagged events start an actor that lives for as long
as the `ActorSystem` does, so call
+`close()` on the cleanup tool once the operations have completed.
+
By default, all operations only print what they were going to do. Once you're
happy with what the cleanup tool is going to do set
`pekko.persistence.cassandra.cleanup.dry-run = false`
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
