<commit_before>package akka.persistence.hbase.journal

import java.util.concurrent.atomic.AtomicBoolean
import java.{util => ju}

import akka.actor.{Actor, ActorLogging, ActorRef, Props}
import akka.persistence.PersistentRepr
import akka.persistence.hbase.common.RowKey
import akka.persistence.hbase.journal.Resequencer.AllPersistentsSubmitted
import akka.persistence.journal._
import org.apache.hadoop.hbase.CellUtil
import org.apache.hadoop.hbase.client.Scan
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp
import org.apache.hadoop.hbase.filter._
import org.apache.hadoop.hbase.util.Bytes
import org.hbase.async.{HBaseClient, KeyValue}

import scala.collection.mutable
import scala.concurrent.{Future, Promise}

trait HBaseAsyncRecovery extends AsyncRecovery {
  this: Actor with ActorLogging with HBaseAsyncWriteJournal =>

  private[persistence] def client: HBaseClient

  private[persistence] implicit def hBasePersistenceSettings: PluginPersistenceSettings

  private lazy val replayDispatcherId = hBasePersistenceSettings.replayDispatcherId

  override implicit val pluginDispatcher = context.system.dispatchers.lookup(replayDispatcherId)

  import akka.persistence.hbase.common.Columns._
  import akka.persistence.hbase.common.DeferredConversions._

import scala.collection.JavaConverters._

  // async recovery plugin impl

  override def asyncReplayMessages(persistenceId: String, fromSequenceNr: Long, toSequenceNr: Long, max: Long)
                                  (replayCallback: PersistentRepr => Unit): Future[Unit] = max match {
    case 0 =>
      log.debug("Skipping async replay for persistenceId [{}], from sequenceNr: [{}], to sequenceNr: [{}], since max messages count to replay is 0",
        persistenceId, fromSequenceNr, toSequenceNr)

      Future.successful() // no need to do a replay anything

    case _ =>
      log.debug("Async replay for persistenceId [{}], from sequenceNr: [{}], to sequenceNr: [{}]{}",
        persistenceId, fromSequenceNr, toSequenceNr, if (max != Long.MaxValue) s", limited to: $max messages" else "")

      val reachedSeqNrPromise = Promise[Long]()
      val loopedMaxFlag = new AtomicBoolean(false) // the resequencer may let us know that it looped `max` messages, and we can abort further scanning
      val resequencer = context.actorOf(Resequencer.props(fromSequenceNr, max, replayCallback, loopedMaxFlag, reachedSeqNrPromise, replayDispatcherId))

      val partitions = hBasePersistenceSettings.partitionCount

      def scanPartition(part: Long, resequencer: ActorRef): Long = {
        val startScanKey = RowKey.firstInPartition(persistenceId, part, fromSequenceNr) // 021-ID-0000000000000000021
        val stopScanKey = RowKey.lastInPartition(persistenceId, part, toSequenceNr) // 021-ID-9223372036854775800
        val persistenceIdRowRegex = RowKey.patternForProcessor(persistenceId) //  .*-ID-.*

        log.info("Scanning {} partition, from {} to {}", part, startScanKey.toKeyString, stopScanKey.toKeyString)

        val scan = new Scan
        scan.setStartRow(startScanKey.toBytes) // inclusive
        scan.setStopRow(stopScanKey.toBytes) // exclusive
        scan.setBatch(hBasePersistenceSettings.scanBatchSize)

        scan.addColumn(FamilyBytes, Marker)
        scan.addColumn(FamilyBytes, Message)

        scan.setFilter(new RowFilter(CompareOp.EQUAL, new RegexStringComparator(persistenceIdRowRegex)))

        val scanner = hTable.getScanner(scan)
        var lowestSeqNr: Long = 0L
        try {
          var res = scanner.next()
          while (res != null) {
            // Note: In case you wonder why we can't break the loop with a simple counter here once we loop through `max` elements:
            // Since this is multiple scans, on multiple partitions, they are not ordered, yet we must deliver ordered messages
            // to the receiver. Only the resequencer knows how many are really "delivered"


            val markerCell = res.getColumnLatestCell(FamilyBytes, Marker)
            val messageCell = res.getColumnLatestCell(FamilyBytes, Message)

            if ((markerCell ne null) && (messageCell ne null)) {
              val marker = Bytes.toString(CellUtil.cloneValue(markerCell))

              marker match {
                case "A" =>
                  val persistentRepr = persistentFromBytes(CellUtil.cloneValue(messageCell))

                  val seqNr = persistentRepr.sequenceNr
                  if (fromSequenceNr <= seqNr && seqNr <= toSequenceNr) {
                    resequencer ! persistentRepr
                    lowestSeqNr = seqNr
                  }

                case "S" =>
                  // thanks to treating Snapshot rows as deleted entries, we won't suddenly apply a Snapshot() where the
                  // our Processor expects a normal message. This is implemented for the HBase backed snapshot storage,
                  // if you use the HDFS storage there won't be any snapshot entries in here.
                  // As for message deletes: if we delete msgs up to seqNr 4, and snapshot was at 3, we want to delete it anyway.
                  // treat as deleted, ignore...

                case "D" =>
                  // marked as deleted, ignore...

                case _ =>
                  // channel confirmation
                  val persistentRepr = persistentFromBytes(CellUtil.cloneValue(messageCell))

                  val channelId = RowTypeMarkers.extractSeqNrFromConfirmedMarker(marker)
                  replayCallback(persistentRepr.update(confirms = channelId +: persistentRepr.confirms))
              }
            }
            res = scanner.next()
          }
          lowestSeqNr
        } finally {
          log.debug("Done scheduling replays in partition {} (lowest seqNr: {})", part, lowestSeqNr)
          scanner.close()
        }
      }
<commit_msg>replaying of marked-as-deleted messages (on demand)<commit_after>package akka.persistence.hbase.journal

import java.util.concurrent.atomic.AtomicBoolean
import java.{util => ju}

import akka.actor.{Actor, ActorLogging, ActorRef, Props}
import akka.persistence.PersistentRepr
import akka.persistence.hbase.common.RowKey
import akka.persistence.hbase.journal.Resequencer.AllPersistentsSubmitted
import akka.persistence.journal._
import org.apache.hadoop.hbase.CellUtil
import org.apache.hadoop.hbase.client.Scan
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp
import org.apache.hadoop.hbase.filter._
import org.apache.hadoop.hbase.util.Bytes
import org.hbase.async.{HBaseClient, KeyValue}

import scala.collection.mutable
import scala.concurrent.{Future, Promise}

trait HBaseAsyncRecovery extends AsyncRecovery {
  this: Actor with ActorLogging with HBaseAsyncWriteJournal =>

  private[persistence] def client: HBaseClient

  private[persistence] implicit def hBasePersistenceSettings: PluginPersistenceSettings

  private lazy val replayDispatcherId = hBasePersistenceSettings.replayDispatcherId

  override implicit val pluginDispatcher = context.system.dispatchers.lookup(replayDispatcherId)

  import akka.persistence.hbase.common.Columns._
  import akka.persistence.hbase.common.DeferredConversions._

import scala.collection.JavaConverters._

  // async recovery plugin impl

  override def asyncReplayMessages(persistenceId: String, fromSequenceNr: Long, toSequenceNr: Long, max: Long)
                                  (replayCallback: PersistentRepr => Unit): Future[Unit] = max match {
    case 0 =>
      log.debug("Skipping async replay for persistenceId [{}], from sequenceNr: [{}], to sequenceNr: [{}], since max messages count to replay is 0",
        persistenceId, fromSequenceNr, toSequenceNr)

      Future.successful() // no need to do a replay anything

    case _ =>
      log.debug("Async replay for persistenceId [{}], from sequenceNr: [{}], to sequenceNr: [{}]{}",
        persistenceId, fromSequenceNr, toSequenceNr, if (max != Long.MaxValue) s", limited to: $max messages" else "")

      val reachedSeqNrPromise = Promise[Long]()
      val loopedMaxFlag = new AtomicBoolean(false) // the resequencer may let us know that it looped `max` messages, and we can abort further scanning
      val resequencer = context.actorOf(Resequencer.props(fromSequenceNr, max, replayCallback, loopedMaxFlag, reachedSeqNrPromise, replayDispatcherId))

      val partitions = hBasePersistenceSettings.partitionCount

      def scanPartition(part: Long, resequencer: ActorRef): Long = {
        val startScanKey = RowKey.firstInPartition(persistenceId, part, fromSequenceNr) // 021-ID-0000000000000000021
        val stopScanKey = RowKey.lastInPartition(persistenceId, part, toSequenceNr) // 021-ID-9223372036854775800
        val persistenceIdRowRegex = RowKey.patternForProcessor(persistenceId) //  .*-ID-.*

        log.info("Scanning {} partition, from {} to {}", part, startScanKey.toKeyString, stopScanKey.toKeyString)

        val scan = new Scan
        scan.setStartRow(startScanKey.toBytes) // inclusive
        scan.setStopRow(stopScanKey.toBytes) // exclusive
        scan.setBatch(hBasePersistenceSettings.scanBatchSize)

        scan.addColumn(FamilyBytes, Marker)
        scan.addColumn(FamilyBytes, Message)

        scan.setFilter(new RowFilter(CompareOp.EQUAL, new RegexStringComparator(persistenceIdRowRegex)))

        val scanner = hTable.getScanner(scan)
        var lowestSeqNr: Long = 0L

        def resequenceMsg(persistentRepr: PersistentRepr) {
          val seqNr = persistentRepr.sequenceNr
          if (fromSequenceNr <= seqNr && seqNr <= toSequenceNr) {
            resequencer ! persistentRepr
            lowestSeqNr = seqNr
          }
        }

        try {
          var res = scanner.next()
          while (res != null) {
            // Note: In case you wonder why we can't break the loop with a simple counter here once we loop through `max` elements:
            // Since this is multiple scans, on multiple partitions, they are not ordered, yet we must deliver ordered messages
            // to the receiver. Only the resequencer knows how many are really "delivered"


            val markerCell = res.getColumnLatestCell(FamilyBytes, Marker)
            val messageCell = res.getColumnLatestCell(FamilyBytes, Message)

            if ((markerCell ne null) && (messageCell ne null)) {
              val marker = Bytes.toString(CellUtil.cloneValue(markerCell))

              marker match {
                case "A" =>
                  val persistentRepr = persistentFromBytes(CellUtil.cloneValue(messageCell))
                  resequenceMsg(persistentRepr)

                case "S" =>
                  // thanks to treating Snapshot rows as deleted entries, we won't suddenly apply a Snapshot() where the
                  // our Processor expects a normal message. This is implemented for the HBase backed snapshot storage,
                  // if you use the HDFS storage there won't be any snapshot entries in here.
                  // As for message deletes: if we delete msgs up to seqNr 4, and snapshot was at 3, we want to delete it anyway.


                case "D" =>
                  // mark as deleted, journal may choose to replay it
                  val persistentRepr = persistentFromBytes(CellUtil.cloneValue(messageCell))
                  resequenceMsg(persistentRepr.update(deleted = true))

                case _ =>
                  // channel confirmation
                  val persistentRepr = persistentFromBytes(CellUtil.cloneValue(messageCell))

                  val channelId = RowTypeMarkers.extractSeqNrFromConfirmedMarker(marker)
                  replayCallback(persistentRepr.update(confirms = channelId +: persistentRepr.confirms))
              }
            }
            res = scanner.next()
          }
          lowestSeqNr
        } finally {
          log.debug("Done scheduling replays in partition {} (lowest seqNr: {})", part, lowestSeqNr)
          scanner.close()
        }
      }

{'fraction_non_alphanumeric': 0.08343783964551459, 'fraction_numerical': 0.00819329487501045, 'mean_word_length': 2.8649434571890144, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 2, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6542414', 'n_tokens_mistral': 3364, 'n_tokens_neox': 3172, 'n_words': 883}