final class Reconciliation extends AnyRef
For reconciling the tag_views table with the messages table. Can be used to:
- Fix data issues caused by split brains or persistence ids running in multiple locations - Continue tag writes after a node crash (using continueTagWritesForPersistenceId) - Rebuild tag views from scratch (using rebuildTagViewForPersistenceIds)
Important: Most operations (delete, truncate, rebuild) should only be run when the affected persistence ids are NOT running. Running these while actors are actively persisting can cause data corruption or gaps in tag sequences. The exception is continueTagWritesForPersistenceId which is safe to run while the application is running.
API likely to change when a java/scaladsl is added.
- Annotations
- @ApiMayChange()
- Source
- Reconciliation.scala
- Alphabetic
- By Inheritance
- Reconciliation
- AnyRef
- Any
- Hide All
- Show All
- Public
- Protected
Instance Constructors
- new Reconciliation(systemProvider: ClassicActorSystemProvider)
- new Reconciliation(systemProvider: ClassicActorSystemProvider, settings: ReconciliationSettings)
Value Members
- final def !=(arg0: Any): Boolean
- Definition Classes
- AnyRef → Any
- final def ##: Int
- Definition Classes
- AnyRef → Any
- final def ==(arg0: Any): Boolean
- Definition Classes
- AnyRef → Any
- def allTags(): Source[String, NotUsed]
Returns all the tags in the journal.
Returns all the tags in the journal. This is not an efficient query for Cassandra so it is better to calculate tags for calls to deleteTagViewForPersistenceId another way.
Prefer to do batches of persistence ids at a time getting the tags just for that persistence id.
- final def asInstanceOf[T0]: T0
- Definition Classes
- Any
- def clone(): AnyRef
- Attributes
- protected[lang]
- Definition Classes
- AnyRef
- Annotations
- @throws(classOf[java.lang.CloneNotSupportedException]) @native()
- def continueTagWritesForPersistenceId(persistenceId: String): Future[Done]
Continues tag writes from where they left off by only writing events that are newer than the current progress (
seqNr > progress.sequenceNr).Continues tag writes from where they left off by only writing events that are newer than the current progress (
seqNr > progress.sequenceNr).Use this when a node crashed or was stopped while tag writes were pending, and you want to complete those writes without the actor having to restart. This is more efficient than rebuildTagViewForPersistenceIds as it:
- Starts reading from the minimum progress sequence number (not from 0) - Only writes events that are actually missing
Concurrency safety: This method can be run while the corresponding actor is running. Tag writes are idempotent because:
- The
timeUuidis from the original event in the main journal (immutable) - Thetag_pid_sequence_nris derived deterministically from the progress baselineIf both the running actor and this tool process the same events, they will calculate the same
tag_pid_sequence_nrvalues, resulting in the same primary key. Cassandra's upsert behavior ensures no duplicates. - def deleteTagViewForPersistenceIds(persistenceId: Set[String], tag: String): Future[Done]
Scans the given tag and deletes all events for the provided persistence ids.
Scans the given tag and deletes all events for the provided persistence ids. All events for a persistence id have to be deleted as not to leave gaps in the tag pid sequence numbers. Also deletes the tag progress for the persistence ids.
Warning: Do not run this while the affected persistence ids are actively running. Concurrent writes from running actors will cause data inconsistencies.
- final def eq(arg0: AnyRef): Boolean
- Definition Classes
- AnyRef
- def equals(arg0: AnyRef): Boolean
- Definition Classes
- AnyRef → Any
- final def getClass(): Class[_ <: AnyRef]
- Definition Classes
- AnyRef → Any
- Annotations
- @native()
- def hashCode(): Int
- Definition Classes
- AnyRef → Any
- Annotations
- @native()
- final def isInstanceOf[T0]: Boolean
- Definition Classes
- Any
- final def ne(arg0: AnyRef): Boolean
- Definition Classes
- AnyRef
- final def notify(): Unit
- Definition Classes
- AnyRef
- Annotations
- @native()
- final def notifyAll(): Unit
- Definition Classes
- AnyRef
- Annotations
- @native()
- def rebuildAllPersistenceIds(): Future[Done]
- def rebuildTagViewForPersistenceIds(persistenceId: String): Future[Done]
Rebuilds the tag view for a persistence id by writing ALL events to the
tag_viewstable.Rebuilds the tag view for a persistence id by writing ALL events to the
tag_viewstable.Important: This method should only be used after deleteTagViewForPersistenceIds or truncateTagView has been called. Running rebuild on existing data will create duplicate entries with different
tag_pid_sequence_nrvalues, causing gaps that will be detected by eventsByTag queries.Warning: Do not run this while the persistence id is actively running. Stop the actor first, then rebuild.
Use this for:
- Rebuilding after deleteTagViewForPersistenceIds or truncateTagView - Initial population when tag writing was previously disabled
For continuing incomplete tag writes (e.g., after a crash), use continueTagWritesForPersistenceId instead.
- final def synchronized[T0](arg0: => T0): T0
- Definition Classes
- AnyRef
- def tagsForPersistenceId(persistenceId: String): Future[Set[String]]
Select all the tags for the given persistence id.
Select all the tags for the given persistence id. This may not return tags that have just been used for the first time.
- def toString(): String
- Definition Classes
- AnyRef → Any
- def truncateTagView(): Future[Done]
Truncate all tag related tables and all metadata so that it can be rebuilt.
Truncate all tag related tables and all metadata so that it can be rebuilt.
Warning: Do not run this while the application is running. Stop all persistent actors first, then truncate, then rebuild or restart the application.
- final def wait(): Unit
- Definition Classes
- AnyRef
- Annotations
- @throws(classOf[java.lang.InterruptedException])
- final def wait(arg0: Long, arg1: Int): Unit
- Definition Classes
- AnyRef
- Annotations
- @throws(classOf[java.lang.InterruptedException])
- final def wait(arg0: Long): Unit
- Definition Classes
- AnyRef
- Annotations
- @throws(classOf[java.lang.InterruptedException]) @native()