Improve performance of deletes over many partitions, #384

core/src/main/scala/akka/persistence/cassandra/journal/CassandraJournal.scala

@@ -403,8 +403,8 @@ class CassandraJournal(cfg: Config) extends AsyncWriteJournal
   }
 
   private def delete(persistenceId: String, toSequenceNr: Long): Future[Unit] = {
-    def physicalDelete(lowestPartition: Long, highestPartition: Long, toSeqNr: Long): Unit = {
-      val partitionInfos = (lowestPartition to highestPartition).map(partitionInfo(persistenceId, _, toSeqNr))
+
+    def physicalDelete(lowestPartition: Long, highestPartition: Long, toSeqNr: Long): Future[Done] = {
       if (config.cassandra2xCompat) {
         def asyncDeleteMessages(partitionNr: Long, messageIds: Seq[MessageId]): Future[Unit] = {
           val boundStatements = messageIds.map(mid =>
@@ -414,62 +414,74 @@ class CassandraJournal(cfg: Config) extends AsyncWriteJournal
           }
         }
 
-        partitionInfos.map(future => future.flatMap(pi => {
-          Future.sequence((pi.minSequenceNr to pi.maxSequenceNr).grouped(config.maxMessageBatchSize).map {
-            group =>
-              {
-                val delete = asyncDeleteMessages(pi.partitionNr, group map (MessageId(persistenceId, _)))
-                delete.failed.foreach { e =>
-                  log.warning(s"Unable to complete deletes for persistence id {}, toSequenceNr {}. " +
-                    "The plugin will continue to function correctly but you will need to manually delete the old messages. " +
-                    "Caused by: [{}: {}]", persistenceId, toSequenceNr, e.getClass.getName, e.getMessage)
+        val partitionInfos = (lowestPartition to highestPartition).map(partitionInfo(persistenceId, _, toSeqNr))
+        val deleteResult =
+          Future.sequence(partitionInfos.map(future => future.flatMap(pi => {
+            Future.sequence((pi.minSequenceNr to pi.maxSequenceNr).grouped(config.maxMessageBatchSize).map {
+              group =>
+                {
+                  val groupDeleteResult = asyncDeleteMessages(pi.partitionNr, group map (MessageId(persistenceId, _)))
+                  groupDeleteResult.failed.foreach { e =>
+                    log.warning(s"Unable to complete deletes for persistence id {}, toSequenceNr {}. " +
+                      "The plugin will continue to function correctly but you will need to manually delete the old messages. " +
+                      "Caused by: [{}: {}]", persistenceId, toSequenceNr, e.getClass.getName, e.getMessage)
+                  }
+                  groupDeleteResult
                 }
-                delete
-              }
-          })
-        }))
+            })
+          })))
+        deleteResult.map(_ => Done)
 
       } else {
-        Future.sequence(partitionInfos.map(future => future.flatMap { pi =>
-          val boundDeleteMessages = preparedDeleteMessages.map(_.bind(persistenceId, pi.partitionNr: JLong, toSeqNr: JLong))
+        val deleteResult = Future.sequence((lowestPartition to highestPartition).map { partitionNr =>
+          val boundDeleteMessages = preparedDeleteMessages.map(_.bind(persistenceId, partitionNr: JLong, toSeqNr: JLong))
           boundDeleteMessages.flatMap(execute(_, deleteRetryPolicy))
-        })).failed.foreach { e =>
+        })
+        deleteResult.failed.foreach { e =>
           log.warning("Unable to complete deletes for persistence id {}, toSequenceNr {}. " +
             "The plugin will continue to function correctly but you will need to manually delete the old messages. " +
             "Caused by: [{}: {}]", persistenceId, toSequenceNr, e.getClass.getName, e.getMessage)
         }
+        deleteResult.map(_ => Done)
       }
     }
 
     /**
-     * Deletes the by inserting into the metadata table deleted_to
+     * Deletes the events by inserting into the metadata table deleted_to
      * and physically deletes the rows.
-     *
-     * The Future completes once the logical delete is complete as
-     * that is when future reads won't see the events.
      */
-    def logicalDeleteWithAsyncPhysicalDelete(lowestSequenceNr: Long, highestSequenceNr: Long): Future[Done] = {
-      val lowestPartition = partitionNr(lowestSequenceNr, config.targetPartitionSize)
+    def logicalAndPhysicalDelete(highestDeletedSequenceNumber: Long, highestSequenceNr: Long): Future[Done] = {
+      val lowestPartition = partitionNr(highestDeletedSequenceNumber + 1, config.targetPartitionSize)
       val toSeqNr = math.min(toSequenceNr, highestSequenceNr)
       val highestPartition = partitionNr(toSeqNr, config.targetPartitionSize) + 1 // may have been moved to the next partition
-      val boundInsertDeletedTo = preparedInsertDeletedTo.map(_.bind(persistenceId, toSeqNr: JLong))
-      val logicalDelete = boundInsertDeletedTo.flatMap(session.executeWrite)
-      // Done async as thw plugin won't see the events once the logical delete is complete
-      logicalDelete.foreach(_ => physicalDelete(lowestPartition, highestPartition, toSeqNr))
-      logicalDelete.map(_ => Done)
+      val logicalDelete =
+        if (toSeqNr <= highestDeletedSequenceNumber) {
+          // already deleted same or higher sequence number, don't update highestDeletedSequenceNumber,
+          // but perform the physical delete (again), may be a retry request
+          Future.successful(())
+        } else {
+          val boundInsertDeletedTo = preparedInsertDeletedTo.map(_.bind(persistenceId, toSeqNr: JLong))
+          boundInsertDeletedTo.flatMap(session.executeWrite)
+        }
+      logicalDelete.flatMap(_ => physicalDelete(lowestPartition, highestPartition, toSeqNr))
     }
 
-    val logicalDelete = for {
+    val deleteResult = for {
       highestDeletedSequenceNumber <- asyncHighestDeletedSequenceNumber(persistenceId)
-      lowestSequenceNr <- asyncReadLowestSequenceNr(persistenceId, 1L, highestDeletedSequenceNumber, Some(config.readConsistency), Some(readRetryPolicy))
-      highestSequenceNr <- asyncFindHighestSequenceNr(persistenceId, lowestSequenceNr, config.targetPartitionSize)
-      _ <- logicalDeleteWithAsyncPhysicalDelete(lowestSequenceNr, highestSequenceNr)
+      highestSequenceNr <- {
+        // MaxValue may be used as magic value to delete all events without specifying actual toSequenceNr
+        if (toSequenceNr == Long.MaxValue)
+          asyncFindHighestSequenceNr(persistenceId, highestDeletedSequenceNumber, config.targetPartitionSize)
+        else
+          Future.successful(toSequenceNr)
+      }
+      _ <- logicalAndPhysicalDelete(highestDeletedSequenceNumber, highestSequenceNr)
     } yield ()
 
-    // Kick off any pending deletes when finished. The physical deltes can work concurrently.
-    logicalDelete.onComplete { result => self ! DeleteFinished(persistenceId, toSequenceNr, result) }
+    // Kick off any pending deletes when finished.
+    deleteResult.onComplete { result => self ! DeleteFinished(persistenceId, toSequenceNr, result) }
 
-    logicalDelete
+    deleteResult
   }
 
   private def partitionInfo(persistenceId: String, partitionNr: Long, maxSequenceNr: Long): Future[PartitionInfo] = {

core/src/test/scala/akka/persistence/cassandra/CassandraLifecycle.scala

@@ -38,6 +38,7 @@ object CassandraLifecycle {
     akka.persistence.snapshot-store.plugin = "cassandra-snapshot-store"
     cassandra-journal.circuit-breaker.call-timeout = 30s
     akka.test.single-expect-default = 20s
+    akka.test.filter-leeway = 20s
     akka.actor.serialize-messages=on
     """
     )

core/src/test/scala/akka/persistence/cassandra/journal/CassandraJournalDeletionSpec.scala

@@ -67,6 +67,22 @@ class CassandraJournalDeletionSpec extends CassandraSpec(
     cassandra-journal-low-concurrent-deletes = ${cassandra-journal}
     cassandra-journal-low-concurrent-deletes {
       max-concurrent-deletes = 5
+      query-plugin = cassandra-query-journal-low-concurrent-deletes
+    }
+    cassandra-query-journal-low-concurrent-deletes = ${cassandra-query-journal}
+    cassandra-query-journal-low-concurrent-deletes {
+      write-plugin = cassandra-journal-low-concurrent-deletes
+    }
+
+    cassandra-journal-small-partition-size = ${cassandra-journal}
+    cassandra-journal-small-partition-size {
+      target-partition-size = 3
+      keyspace = "DeletionSpecMany"
+      query-plugin = cassandra-query-journal-small-partition-size
+    }
+    cassandra-query-journal-small-partition-size = ${cassandra-query-journal}
+    cassandra-query-journal-small-partition-size {
+      write-plugin = cassandra-journal-small-partition-size
     }
   """
 ) {
@@ -126,5 +142,43 @@ class CassandraJournalDeletionSpec extends CassandraSpec(
       }
       successes shouldEqual successes.sorted
     }
+
+    "handle deletes over many partitions" in {
+      val deleteSuccess = TestProbe()
+      val deleteFail = TestProbe()
+      val props = Props(new PAThatDeletes("p3", deleteSuccess.ref, deleteFail.ref, "cassandra-journal-small-partition-size"))
+      val p1 = system.actorOf(props)
+      (1 to 100).foreach { i =>
+        p1 ! PersistMe(i)
+        expectMsg(Ack)
+      }
+
+      p1 ! DeleteTo(10)
+      deleteSuccess.expectMsg(Deleted(10))
+
+      p1 ! DeleteTo(20)
+      deleteSuccess.expectMsg(Deleted(20))
+
+      p1 ! DeleteTo(98)
+      deleteSuccess.expectMsg(Deleted(98))
+
+      p1 ! PoisonPill
+
+      // Recovery should not find a deleted sequence nr
+      val p1TakeTwo = system.actorOf(props)
+      p1TakeTwo ! GetRecoveredEvents
+      expectMsg(RecoveredEvents(List(PersistMe(99), PersistMe(100), RecoveryCompleted)))
+
+      // Delete all with Long.MaxValue
+      p1TakeTwo ! DeleteTo(Long.MaxValue)
+      deleteSuccess.expectMsg(Deleted(Long.MaxValue))
+
+      p1TakeTwo ! PoisonPill
+
+      val p1TakeThree = system.actorOf(props)
+      p1TakeThree ! GetRecoveredEvents
+      expectMsg(RecoveredEvents(List(RecoveryCompleted)))
+    }
+
   }
 }

core/src/test/scala/akka/persistence/cassandra/journal/MultiPluginSpec.scala

@@ -22,6 +22,7 @@ object MultiPluginSpec {
   val config = ConfigFactory.parseString(
     s"""
        |akka.test.single-expect-default = 20s
+       |akka.test.filter-leeway = 20s
        |
        |cassandra-journal.keyspace = $journalKeyspace
        |cassandra-journal.keyspace-autocreate=false