Initial commit

Author: Marc Rooding

build.sbt

@@ -0,0 +1,39 @@
+
+name := "akka-streams-kafka-last-n-minutes"
+
+version := "0.1"
+
+scalaVersion := "2.13.1"
+
+scalacOptions := Seq(
+  "-encoding", "utf8",
+  "-target:jvm-1.8",
+  "-feature",
+  "-language:implicitConversions",
+  "-language:postfixOps",
+  "-unchecked",
+  "-deprecation",
+  "-Xlog-reflective-calls"
+)
+
+resolvers ++= Seq(
+  Resolver.mavenCentral
+)
+
+val akkaVersion = "2.5.23"
+val circeVersion = "0.12.3"
+
+libraryDependencies ++= Seq(
+  "com.typesafe.akka"           %% "akka-actor"                  % akkaVersion,
+  "com.typesafe.akka"           %% "akka-stream"                 % akkaVersion,
+  "com.typesafe.akka"           %% "akka-stream-kafka"           % "1.0.4",
+
+  "io.circe"                    %% "circe-core"                  % circeVersion,
+  "io.circe"                    %% "circe-generic"               % circeVersion,
+  "io.circe"                    %% "circe-parser"                % circeVersion,
+
+  "com.typesafe.scala-logging"  %% "scala-logging"               % "3.9.2",
+
+  "com.typesafe.akka"           %% "akka-stream-testkit"         % akkaVersion       % Test,
+  "org.scalatest"               %% "scalatest"                   % "3.0.8"           % Test,
+)

project/build.properties

@@ -0,0 +1 @@
+sbt.version = 1.3.3

src/main/resources/akka-kafka.conf

@@ -0,0 +1,5 @@
+akka.kafka.consumer {
+  kafka-clients {
+    enable.auto.commit = false
+  }
+}

src/main/resources/application.conf

@@ -0,0 +1,12 @@
+me.mrooding.blog {
+  kafka {
+    group-id: ${?KAFKA_GROUPID}
+
+    broker-list: ${?KAFKA_BROKERLIST}
+
+    topics {
+      price: ${?KAFKA_PRICE_TOPIC}
+    }
+  }
+}
+

src/main/scala/me/mrooding/blog/AppConfig.scala

@@ -0,0 +1,33 @@
+package me.mrooding.blog
+
+import akka.kafka.ConsumerSettings
+import com.typesafe.config.{Config, ConfigFactory}
+import me.mrooding.blog.AppConfig.{KafkaConfig, KafkaTopics}
+import org.apache.kafka.clients.consumer.ConsumerConfig
+import org.apache.kafka.common.serialization.{ByteArrayDeserializer, StringDeserializer}
+
+class AppConfig(
+                  config: Config = ConfigFactory.load().getConfig("me.mrooding.blog"),
+                  akkaKafkaConfig: Config = ConfigFactory.load().getConfig("akka.kafka")
+                ) {
+  val kafka = KafkaConfig(
+    groupId = config.getString("kafka.group-id"),
+    brokerList = config.getString("kafka.broker-list"),
+    topics = KafkaTopics(
+      price = config.getString("kafka.topics.price"),
+    ),
+    bufferSize = config.getInt("kafka.producer.buffer-size")
+  )
+
+  val consumerSettings: ConsumerSettings[String, Array[Byte]] =
+    ConsumerSettings(akkaKafkaConfig.getConfig("consumer"), new StringDeserializer, new ByteArrayDeserializer)
+      .withBootstrapServers(kafka.brokerList)
+      .withGroupId(kafka.groupId)
+      .withProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest")
+}
+
+object AppConfig {
+  case class KafkaTopics(price: String)
+  case class KafkaConfig(groupId: String, brokerList: String, topics: KafkaTopics, bufferSize: Int)
+}
+

src/main/scala/me/mrooding/blog/Main.scala

@@ -0,0 +1,45 @@
+package me.mrooding.blog
+
+import java.time.{Duration, Instant}
+
+import akka.actor.ActorSystem
+import akka.kafka.KafkaConsumerActor
+import akka.stream.ActorMaterializer
+import com.typesafe.scalalogging.LazyLogging
+import me.mrooding.blog.kafka.{KafkaPartitionsResolver, KafkaPriceConsumerService}
+import me.mrooding.blog.publish.PriceUpdater
+
+import scala.concurrent.ExecutionContextExecutor
+import scala.util.{Failure, Success, Try}
+
+object Main extends App with LazyLogging {
+  implicit val system: ActorSystem = ActorSystem("mune")
+  implicit val materializer: ActorMaterializer = ActorMaterializer()
+  implicit val executionContext: ExecutionContextExecutor = system.dispatcher
+
+  Try(new AppConfig()) match {
+    case Success(config) =>
+      val kafkaConsumer = system.actorOf(KafkaConsumerActor.props(config.consumerSettings))
+      val kafkaPartitionsResolver = new KafkaPartitionsResolver(kafkaConsumer)
+
+      kafkaPartitionsResolver
+        .retrievePartitionsForTopic(config.kafka.topics.price)
+        .flatMap { partitions =>
+          val priceUpdater = system.actorOf(PriceUpdater.props())
+          val kafkaRFQConsumerService = new KafkaPriceConsumerService(config, partitions, priceUpdater)
+
+          val timeHorizon = Duration.ofMinutes(5)
+          val horizonStart = Instant.now().minus(timeHorizon)
+
+          val consumerServiceDone = kafkaRFQConsumerService.start(horizonStart)
+
+          consumerServiceDone
+            .andThen {
+              case Success(_) =>
+                logger.info("RFQ consumer service has stopped.")
+              case Failure(exception) =>
+                logger.error("RFQ consumer service has stopped due to error.", exception)
+            }
+        }
+  }
+}

src/main/scala/me/mrooding/blog/data/Price.scala

@@ -0,0 +1,5 @@
+package me.mrooding.blog.data
+
+import java.time.Instant
+
+case class Price(isin: String, price: BigDecimal, timestamp: Instant)

src/main/scala/me/mrooding/blog/json/PriceProtocol.scala

@@ -0,0 +1,11 @@
+package me.mrooding.blog.json
+
+import io.circe.Decoder
+import io.circe.generic.semiauto
+import me.mrooding.blog.data.Price
+
+trait PriceProtocol {
+  implicit val priceFromKafkaJson: Decoder[Price] = semiauto.deriveDecoder
+}
+
+object PriceProtocol extends PriceProtocol

src/main/scala/me/mrooding/blog/kafka/KafkaPartitionsResolver.scala

@@ -0,0 +1,22 @@
+package me.mrooding.blog.kafka
+
+import akka.actor.ActorRef
+import akka.kafka.Metadata
+import akka.kafka.Metadata.GetPartitionsFor
+import akka.pattern._
+import akka.util.Timeout
+import org.apache.kafka.common.PartitionInfo
+
+import scala.concurrent.ExecutionContext.Implicits.global
+import scala.concurrent.Future
+import scala.concurrent.duration._
+
+class KafkaPartitionsResolver(consumer: ActorRef) {
+  implicit val timeout: Timeout = Timeout(5 seconds)
+
+  def retrievePartitionsForTopic(topic: String): Future[List[PartitionInfo]] = {
+    val partitionsFor = (consumer ? GetPartitionsFor(topic)).mapTo[Metadata.PartitionsFor]
+
+    partitionsFor.flatMap(p => Future.fromTry(p.response))
+  }
+}

src/main/scala/me/mrooding/blog/kafka/KafkaPriceConsumerService.scala

@@ -0,0 +1,95 @@
+package me.mrooding.blog.kafka
+
+import java.nio.charset.StandardCharsets
+import java.time.Instant
+
+import akka.actor.{ActorRef, ActorSystem}
+import akka.kafka.Subscriptions
+import akka.kafka.scaladsl.Consumer
+import akka.kafka.scaladsl.Consumer.Control
+import akka.pattern._
+import akka.stream.Materializer
+import akka.stream.scaladsl.{MergeLatest, Sink, Source}
+import akka.util.Timeout
+import akka.{Done, NotUsed}
+import com.typesafe.scalalogging.LazyLogging
+import io.circe.parser.decode
+import me.mrooding.blog.AppConfig
+import me.mrooding.blog.data.Price
+import me.mrooding.blog.publish.PriceUpdater.UpdatePrice
+import org.apache.kafka.clients.consumer.ConsumerRecord
+import org.apache.kafka.common.{PartitionInfo, TopicPartition}
+
+import scala.collection.SortedMap
+import scala.concurrent.duration._
+import scala.concurrent.{ExecutionContext, Future}
+
+class KafkaPriceConsumerService(config: AppConfig, partitions: Seq[PartitionInfo], priceUpdater: ActorRef)(
+  implicit
+  val system: ActorSystem,
+  implicit val materializer: Materializer
+) extends LazyLogging {
+  import KafkaPriceConsumerService._
+
+  require(partitions.nonEmpty, "Cannot consume without any partitions")
+
+  def start(timeHorizon: Instant): Future[Done] = {
+    val consumerSettings = config.consumerSettings
+    val kafkaSources: SortedMap[Partition, Source[ConsumerRecord[String, Array[Byte]], Control]] =
+      SortedMap(partitions.map { partition =>
+        val topicPartition = new TopicPartition(partition.topic(), partition.partition())
+        val partitionSubscription = Subscriptions.assignmentOffsetsForTimes(topicPartition, timeHorizon.toEpochMilli)
+        Partition(partition.partition()) -> Consumer.plainSource[String, Array[Byte]](consumerSettings, partitionSubscription)
+      }: _*)
+
+    implicit val timeout: Timeout = Timeout(10 seconds)
+    val publisher = jsonPublisher(rfq => priceUpdater ? UpdatePrice(rfq), parallelism = 1)(system.dispatcher)
+    val rfqPublishers: Seq[Source[(Partition, RecordPosition), Control]] = kafkaSources.map {
+      case (partition, source) =>
+        source.via(publisher).map(position => partition -> position)
+    }.toSeq
+
+    val compositeStream: Source[List[(Partition, RecordPosition)], _] = rfqPublishers match {
+      case Seq(onlyPartition)            => onlyPartition.map(x => List(x))
+      case Seq(first, second)            => Source.combine(first, second)(MergeLatest(_, eagerComplete = true))
+      case Seq(first, second, rest @ _*) => Source.combine(first, second, rest: _*)(MergeLatest(_, eagerComplete = true))
+    }
+    compositeStream
+      .log(getClass.getSimpleName)
+      .runWith(Sink.ignore)
+  }
+}
+
+object KafkaPriceConsumerService extends LazyLogging {
+  import akka.stream.scaladsl._
+
+  final case class Partition(id: Long) extends AnyVal
+
+  object Partition {
+    implicit def orderingById: Ordering[Partition] = Ordering.by(p => p.id)
+  }
+  final case class RecordPosition(offset: Long, timestamp: Instant)
+
+  def jsonPublisher(publish: Price => Future[Any], parallelism: Int)(
+    implicit executionContext: ExecutionContext
+  ): Flow[ConsumerRecord[_, Array[Byte]], RecordPosition, NotUsed] =
+    Flow[ConsumerRecord[_, Array[Byte]]]
+      .map { record =>
+        import me.mrooding.blog.json.PriceProtocol.priceFromKafkaJson
+
+        val message = new String(record.value(), StandardCharsets.UTF_8)
+        val decodingResult = decode(message)
+
+        decodingResult.left
+          .foreach(decodingFailure => logger.warn(s"Failed to parse ConsumerRecord to Price: $message", decodingFailure))
+
+        RecordPosition(record.offset(), Instant.ofEpochMilli(record.timestamp())) -> decodingResult
+      }
+      .mapAsync(parallelism) {
+        case (position, Right(rfq)) =>
+          publish(rfq).map(_ => position)
+        case (position, _) =>
+          /* We logged the decoding failure above. We'll just skip the unprocessable RFQ for now */
+          Future.successful(position)
+      }
+}

src/main/scala/me/mrooding/blog/publish/PriceUpdater.scala

@@ -0,0 +1,26 @@
+package me.mrooding.blog.publish
+
+import akka.Done
+import akka.actor.{Actor, ActorLogging, ActorRef, ActorSystem, Props}
+import me.mrooding.blog.data.Price
+import me.mrooding.blog.publish.PriceUpdater.UpdatePrice
+
+import scala.concurrent.ExecutionContext
+
+class PriceUpdater()(implicit system: ActorSystem, execContext: ExecutionContext)
+  extends Actor
+    with ActorLogging {
+  override def receive: Receive = {
+    case UpdatePrice(price) =>
+      log.debug(s"Updating price: $price")
+      sender() ! Done
+  }
+}
+
+object PriceUpdater {
+
+  case class UpdatePrice(price: Price)
+
+  def props()(implicit system: ActorSystem, context: ExecutionContext) =
+    Props(new PriceUpdater())
+}