Schedule messages

avro/src/main/scala/GenerateSchema.scala

@@ -2,7 +2,7 @@ import java.nio.charset.StandardCharsets
 import java.nio.file.{Files, Paths}
 
 import com.sksamuel.avro4s._
-import com.sky.kafka.message.scheduler.domain.Schedule
+import com.sky.kms.domain.Schedule
 
 object GenerateSchema extends App {
 

build.sbt

@@ -9,22 +9,28 @@ val dependencies = Seq(
   "com.typesafe.akka"          %% "akka-stream"                % akkaVersion,
   "com.typesafe.akka"          %% "akka-slf4j"                 % akkaVersion,
   "com.typesafe.akka"          %% "akka-stream-kafka"          % "0.16",
+  "com.typesafe.akka"          %% "akka-stream-contrib"        % "0.8",
 
   "com.typesafe.scala-logging" %% "scala-logging"              % "3.5.0",
   "com.sksamuel.avro4s"        %% "avro4s-core"                % "1.7.0",
   "org.typelevel"              %% "cats"                       % "0.9.0",
   "ch.qos.logback"              % "logback-classic"            % "1.2.3"      % Runtime,
   "com.github.pureconfig"      %% "pureconfig"                 % "0.7.2",
+  "org.zalando"                %% "grafter"                    % "2.0.1",
 
   "io.kamon"                   %% "kamon-jmx"                  % kamonVersion,
   "io.kamon"                   %% "kamon-akka-2.5"             % kamonVersion,
   "io.kamon"                   %% "kamon-core"                 % kamonVersion,
 
   "org.scalatest"              %% "scalatest"                  % "3.0.1"      % Test,
   "com.typesafe.akka"          %% "akka-testkit"               % akkaVersion  % Test,
+  "com.typesafe.akka"          %% "akka-stream-testkit"        % akkaVersion  % Test,
   "net.cakesolutions"          %% "scala-kafka-client-testkit" % kafkaVersion % Test,
   "org.slf4j"                   % "log4j-over-slf4j"           % "1.7.21"     % Test,
-  "com.danielasfregola"        %% "random-data-generator"      % "2.1"        % Test
+  "com.danielasfregola"        %% "random-data-generator"      % "2.1"        % Test,
+  "com.47deg"                  %% "scalacheck-toolbox-datetime"% "0.2.2"      % Test,
+  "com.miguno.akka"            %% "akka-mock-scheduler"        % "0.5.1"      % Test,
+  "org.mockito"                 % "mockito-all"                % "1.10.19"    % Test
 )
 
 val commonSettings = Seq(
@@ -73,7 +79,8 @@ lazy val scheduler = (project in file("scheduler"))
     javaAgents += "org.aspectj" % "aspectjweaver" % "1.8.10",
     javaOptions in Universal += jmxSettings,
     buildInfoSettings,
-    dockerSettings
+    dockerSettings,
+    dependencyOverrides += "org.scalacheck" %% "scalacheck" % "1.13.5"
   ).enablePlugins(DockerPlugin)
 
 val schema = inputKey[Unit]("Generate the Avro schema file for the Schedule schema.")

scheduler/src/main/resources/application.conf

@@ -2,10 +2,6 @@ scheduler {
   schedule-topic = ${?SCHEDULE_TOPIC}
   kafka-brokers = "localhost:9092"
   kafka-brokers = ${?KAFKA_BROKERS}
-  shutdown-timeout {
-    stream = 10 seconds
-    system = 10 seconds
-  }
 }
 
 akka {
@@ -25,7 +21,8 @@ akka {
       use-dispatcher = "akka.kafka.default-dispatcher"
 
       kafka-clients {
-        enable.auto.commit = false
+        // until we use a commitableSource
+        enable.auto.commit = true
 
         group.id = "com.sky.kafka.scheduler"
         bootstrap.servers = ${scheduler.kafka-brokers}

scheduler/src/main/resources/reference.conf

@@ -0,0 +1,7 @@
+scheduler {
+  shutdown-timeout {
+    stream = 10 seconds
+    system = 10 seconds
+  }
+  queue-buffer-size = 100
+}

scheduler/src/main/scala/com/sky/kms/AkkaComponents.scala

@@ -0,0 +1,23 @@
+package com.sky.kms
+
+import akka.actor.ActorSystem
+import akka.stream.Supervision.Restart
+import akka.stream.{ActorMaterializer, ActorMaterializerSettings, Supervision}
+import com.typesafe.scalalogging.LazyLogging
+
+trait AkkaComponents extends LazyLogging with Monitoring {
+
+  implicit val system = ActorSystem("kafka-message-scheduler")
+
+  val decider: Supervision.Decider = { t =>
+    recordException(t)
+    logger.error(s"Supervision failed.", t)
+    Restart
+  }
+
+  val settings = ActorMaterializerSettings(system)
+    .withSupervisionStrategy(decider)
+
+  implicit val materializer = ActorMaterializer(settings)
+
+}

scheduler/src/main/scala/com/sky/kms/Monitoring.scala

@@ -0,0 +1,20 @@
+package com.sky.kms
+
+import kamon.Kamon
+import kamon.metric.MetricsModule
+
+trait Monitoring {
+
+  val metrics: MetricsModule = Kamon.metrics
+
+  def increment(key: String) = metrics.counter(key).increment()
+
+  def recordException(throwable: Throwable) = {
+    val key = generateKeyFromException(throwable)
+    metrics.counter(key).increment()
+  }
+
+  private def generateKeyFromException(throwable: Throwable): String = {
+    return s"exception.${throwable.getClass.getName.replace(".", "_")}"
+  }
+}

scheduler/src/main/scala/com/sky/kms/SchedulerApp.scala

@@ -0,0 +1,29 @@
+package com.sky.kms
+
+import com.sky.kms.config.AppConfig
+import com.sky.kms.streams.ScheduleReader
+import com.typesafe.scalalogging.LazyLogging
+import kamon.Kamon
+import org.zalando.grafter._
+import pureconfig._
+
+import scala.concurrent.Await
+
+object SchedulerApp extends App with LazyLogging with AkkaComponents {
+
+  val conf = loadConfigOrThrow[AppConfig]
+  Kamon.start()
+
+  logger.info("Kafka Message Scheduler starting up...")
+  val app = ScheduleReader.reader.run(conf)
+
+  sys.addShutdownHook {
+    logger.info("Kafka Message Scheduler shutting down...")
+    Rewriter.stop(app).value
+
+    materializer.shutdown()
+    Await.result(system.terminate(), conf.scheduler.shutdownTimeout.system)
+
+    Kamon.shutdown()
+  }
+}