<commit_before>package io.flow.event.actors

import akka.actor.{Actor, ActorLogging, ActorSystem}
import io.flow.event.{Queue, MockQueue, Record}
import io.flow.play.actors.ErrorHandler
import play.api.Logger

import scala.concurrent.ExecutionContext
import scala.concurrent.duration._
import scala.reflect.runtime.universe.TypeTag
import scala.util.{Failure, Success, Try}

/**
  * Poll Actor periodically polls a kinesis stream (by default every 5
  * seconds), invoking process once per message.
  * 
  * To extend this class:
  *   - implement system, queue, process(record)
  *   - call start(...) w/ the name of the execution context to use
  */
trait PollActor extends Actor with ActorLogging with ErrorHandler {

  /**
    * Called once for every event read off the stream
    */
  def process(record: Record)

  def system: ActorSystem

  def queue: Queue

  private[this] implicit var ec: ExecutionContext = null

  private[this] def defaultDuration = {
    queue match {
      case q:  MockQueue => FiniteDuration(10, MILLISECONDS)
      case _ => FiniteDuration(5, SECONDS)
    }
  }

  def start[T: TypeTag](
    executionContextName: String,
    pollTime: FiniteDuration = defaultDuration
  ) {
    val ec = system.dispatchers.lookup(executionContextName)
    startWithExecutionContext(ec, pollTime)
  }

  def startWithExecutionContext[T: TypeTag](
    executionContext: ExecutionContext,
    pollTime: FiniteDuration = FiniteDuration(5, SECONDS)
  ) {
    Logger.info(s"[${getClass.getName}] Scheduling poll every $pollTime")

    this.ec = executionContext
    this.stream = Some(queue.stream[T])

    system.scheduler.schedule(pollTime, pollTime, self, Poll)
  }

  private[this] var stream: Option[io.flow.event.Stream] = None
  private[this] case object Poll

  def receive = akka.event.LoggingReceive {

    case msg @ Poll => withErrorHandler(msg) {
      stream match {
        case None => {
          sys.error("Must call start before polling")
        }

        case Some(s) => {
          s.consume { record =>
            Try {
              process(record)
            } match {
              case Success(_) => // no-op
              case Failure(ex) => {
                ex.printStackTrace(System.err)

                // explicitly catch and only warn on duplicate key value constraint errors on partitioned tables
                if (PollActor.filterExceptionMessage(ex.getMessage)) {
                  Logger.warn(s"[${self.getClass.getName}] FlowEventWarning Error processing record: ${ex.getMessage}")
                } else {
                  Logger.error(s"[${self.getClass.getName}] FlowEventError Error processing record: ${ex.getMessage}", ex)
                }
              }
            }
          }
        }
      }
    }

    case msg: Any => logUnhandledMessage(msg)

  }
}

object PollActor {
  /** Checks whether the first line of an exception message matches a partman partitioning error, which is not critical. */
  def filterExceptionMessage(message: String): Boolean = {
    message.split("\\r?\\n").headOption.exists(_.matches(".*duplicate key value violates unique constraint.*_p\\d{4}_\\d{2}_\\d{2}_pkey.*"))
  }
}
<commit_msg>sys.error on process failure (#212)<commit_after>package io.flow.event.actors

import akka.actor.{Actor, ActorLogging, ActorSystem}
import io.flow.event.{Queue, MockQueue, Record}
import io.flow.play.actors.ErrorHandler
import play.api.Logger

import scala.concurrent.ExecutionContext
import scala.concurrent.duration._
import scala.reflect.runtime.universe.TypeTag
import scala.util.{Failure, Success, Try}

/**
  * Poll Actor periodically polls a kinesis stream (by default every 5
  * seconds), invoking process once per message.
  * 
  * To extend this class:
  *   - implement system, queue, process(record)
  *   - call start(...) w/ the name of the execution context to use
  */
trait PollActor extends Actor with ActorLogging with ErrorHandler {

  /**
    * Called once for every event read off the stream
    */
  def process(record: Record)

  def system: ActorSystem

  def queue: Queue

  private[this] implicit var ec: ExecutionContext = null

  private[this] def defaultDuration = {
    queue match {
      case q:  MockQueue => FiniteDuration(10, MILLISECONDS)
      case _ => FiniteDuration(5, SECONDS)
    }
  }

  def start[T: TypeTag](
    executionContextName: String,
    pollTime: FiniteDuration = defaultDuration
  ) {
    val ec = system.dispatchers.lookup(executionContextName)
    startWithExecutionContext(ec, pollTime)
  }

  def startWithExecutionContext[T: TypeTag](
    executionContext: ExecutionContext,
    pollTime: FiniteDuration = FiniteDuration(5, SECONDS)
  ) {
    Logger.info(s"[${getClass.getName}] Scheduling poll every $pollTime")

    this.ec = executionContext
    this.stream = Some(queue.stream[T])

    system.scheduler.schedule(pollTime, pollTime, self, Poll)
  }

  private[this] var stream: Option[io.flow.event.Stream] = None
  private[this] case object Poll

  def receive = akka.event.LoggingReceive {

    case msg @ Poll => withErrorHandler(msg) {
      stream match {
        case None => {
          sys.error("Must call start before polling")
        }

        case Some(s) => {
          s.consume { record =>
            Try {
              process(record)
            } match {
              case Success(_) => // no-op
              case Failure(ex) => {
                ex.printStackTrace(System.err)

                sys.error(s"[${self.getClass.getName}] FlowEventError Error processing record: ${ex.getMessage}")
              }
            }
          }
        }
      }
    }

    case msg: Any => logUnhandledMessage(msg)

  }
}

object PollActor {
  /** Checks whether the first line of an exception message matches a partman partitioning error, which is not critical. */
  def filterExceptionMessage(message: String): Boolean = {
    message.split("\\r?\\n").headOption.exists(_.matches(".*duplicate key value violates unique constraint.*_p\\d{4}_\\d{2}_\\d{2}_pkey.*"))
  }
}

{'fraction_non_alphanumeric': 0.10728476821192053, 'fraction_numerical': 0.0031456953642384107, 'mean_word_length': 2.796983029541169, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5203530', 'n_tokens_mistral': 1790, 'n_tokens_neox': 1716, 'n_words': 464}