Functional Programming with Kafka Streams and Scala by Joan Goyeau

Our main requirement is that the system should scale horizontally on reads and writes.

Read optimised approach

Write optimised approach

Another way of doing is probably doing some denormalisation at write time. So we could create a DenormalisedPost case class that will contain all the necessary informations for the reads:

case class DenormalisedPost(post: Post, author: User, interactions: Interactions)
case class Interactions(likes: Set[Like], comments: Int)

And then we could at write time update this DenormalisedPost, for example incrementing the like counter when someone like the post.
This would make the reads very fast but it wouldn’t scale if we have many writes. For example if a post is very popular and so many people rush to like it, we would be locking the denormalised post way too much.

CQRS approach

A real solution is to implement the CQRS (Command Query Responsibility Segregation) model:

This basically separates the reads and the writes to let them scale. We would therefore be able from the API to write at high throughput new users, posts, comments and likes in the command model and read the denormalised posts in the query model.

One thing is missing though, if we separate the reads and the writes how the data we write in the command model will end up in the query model?
This is what Kafka Streams is for, it’s a data processing library that can consume data from topics, transform them and sink in other topics.

So here is how we can implement the CQRS approach with Kafka and Kafka Streams:

The API will write it’s commands in some Kafka topics. Those topics will be consumed by Kafka Streams, transformed and sank in some other topics. Those other topics will be connected to write in databases like Elasticsearch, MongoDB, Redis… whichever is the best for the API to query.

Note that this architecture is eventually consistent, this means if I like a post and query straight away the number of likes on this post it might not already be incremented. We usually solve this issue with the frontend faking the increment in the counter.

Command Model

• users
• posts
• likes
• comments

So the API will be able to write the case classes previously defined in their corresponding topics:

import com.goyeau.kafka.streams.circe.CirceSerdes._

val config = new Properties()
config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092")
val postProducer = new KafkaProducer(
  config,
  CirceSerdes.serializer[Id[Post]],
  CirceSerdes.serializer[Post]
)

val post = Post(Id[Post]("post0"), Instant.now(), Id[User]("user0"), "Some text", URI.create("https://some-uri"), false)
producer.send(new ProducerRecord("posts", null, post.updatedOn, post.id, post)).get()
postProducer.close()
view rawProducer.scala hosted with ❤ by GitHub

trait Record[K, V] {
  def topic: String
  def key(value: V): K
  def timestamp(value: V): Long
}

object Producer {
  def apply[V] = new ProducerBuilder[V]
  class ProducerBuilder[V] {
    def apply[K](config: Properties)(implicit record: Record[K, V],
                                     keySerializer: Serializer[K],
                                     valueSerializer: Serializer[V]): KafkaProducer[K, V] =
      new KafkaProducer(config, keySerializer, valueSerializer)
  }
}

implicit class KafkaProducerOps[K, V](kafkaProducer: KafkaProducer[K, V]) {
  def send(value: V)(implicit record: Record[K, V]): Future[RecordMetadata] = Future {
    kafkaProducer.send(new ProducerRecord(record.topic, null, record.timestamp(value), record.key(value), value)).get()
  }
}

object Post {
  implicit val record: Record[Id[Post], Post] = new Record[Id[Post], Post] {
    val topic = "posts"
    def key(post: Post): Id[Post] = post.id
    def timestamp(post: Post): Long = post.updatedOn.toEpochMilli
  }
}

val streamsBuilder = new StreamsBuilder()
val postsStream = streamsBuilder.stream[Id[Post], Post]("posts")

implicit class StreamsBuilderOps(streamsBuilder: StreamsBuilderS) {
  def streamFromRecord[V] = new StreamBuilder[V]

  class StreamBuilder[V] {
    def apply[K]()(implicit record: Record[K, V], consumed: Consumed[K, V]): KStreamS[K, V] =
      streamsBuilder.stream[K, V](record.topic)
  }
}

val usersStream = streamsBuilder.streamFromRecord[User]()
val postsStream = streamsBuilder.streamFromRecord[Post]()
val commentsStream = streamsBuilder.streamFromRecord[Comment]()
val likesStream = streamsBuilder.streamFromRecord[Like]()

Alright, you’ve been reading for too long now to finally see the first stream processing code that will power our social network, be carful Mark Zuckerberg!
The goal here is to generate DenormalisedPost as we receive Users, Posts, Likes and Comments. For that we will need to build KTables that will represent our state and then we will be able to join them on their common key.

So we want a KTable of users by their id to join with a KTable of the posts by author id (user id).

You might wonder why using a KTable for posts? Why not using the original KStream rekeyed by author id to join with the KTable of users by their id, essentially querying a user for each post that comes?
Well for the reason that if a user gets updated after he posted, we will need to find back all the posts he created to update the user there too. This can only be done if we keep the posts in a KTable, effectively building a queryable state for them.

Commutativity

This brings us on the concept of commutativity. A simple example is:

1 + 2 + 3 = 6
3 + 2 + 1= 6

Therefore we can say that the + operator is commutative since it doesn’t matter in which order the numbers are coming.

This is exactly what we are doing here with the Posts and Users, it shouldn’t matter in which order they are coming, the output should always be the same DenormalisedPost.

Alright, let’s have a look at the code that creates these 2 KTables:

val usersByKey = usersStream
  .groupByKey
  .reduce((first, second) => if (first.updatedOn.isAfter(second.updatedOn)) first else second)

val postsByAuthor = postsStream
  .groupBy((_, post) => post.author)
  .aggregate(
    () => Map.empty[Id[Post], Post],
    (_, post: Post, posts: Map[Id[Post], Post]) =>
      if (posts.get(post.id).exists(_.updatedOn.isAfter(post.updatedOn))) posts
      else posts + (post.id -> post),
    Materialized.`with`[Id[User], Map[Id[Post], Post], KeyValueStore[Bytes, Array[Byte]]](
      CirceSerdes.serde[Id[User]],
      CirceSerdes.serde[Map[Id[Post], Post]]
    )
  )
  .mapValues(_.values.toSet)

• usersByKey group by the key (user id) and reduce all the user we get for the same key to keep only the latest version of it. The check on isAfterensure commutativity, as it doesn’t matter in which order the reduce receives the users, the result will be the same.
• postsByAuthor group by the author of the post (user id) and aggregate all the posts we get for the same author id. We are also doing the check on isAfter to stay commutative. One important point here is that we are grouping, not by key, but by another field of the document (the author), this operation is more expensive than grouping by key as it envolves repartitioning of the stream. Although this is required here as we will see later, the 2 KTables or KStreams we are joining need to be co-partitioned.

Associativity

It is important here that we also respect the concept of associativity. For example the + and * operators used together are not associative:

(3*5)+6 = 21
3*(5+6) = 33

As we can see the result will not be the same depending how you apply the operators even though the order of the elements is the same.

If we look at the reduce operation of usersByKey, it doesn’t matter how we apply it in regards to their updatedOn field:

(2018–05–20, 2018–05–01), 2018–05–28 = 2018–05–28
2018–05–20, (2018–05–01, 2018–05–28) = 2018–05–28

The result will always be the same.

Let’s add the KTables for Likes and Comments:

val likesByKey = likesStream
  .groupByKey
  .aggregate(
    () => Set.empty[Like],
    (_, like: Like, likes: Set[Like]) => if (like.unliked) likes - like else likes + like,
    Materialized.`with`[Id[Post], Set[Like], KeyValueStore[Bytes, Array[Byte]]](
      CirceSerdes.serde[Id[Post]],
      CirceSerdes.serde[Set[Like]]
    )
  )

val commentCountByKey = commentsStream
  .groupByKey
  .aggregate(
    () => 0,
    (_, comment: Comment, count: Int) => if (comment.deleted) count - 1 else count + 1,
    Materialized.`with`[Id[Post], Int, KeyValueStore[Bytes, Array[Byte]]](
      CirceSerdes.serde[Id[Post]],
      CirceSerdes.serde[Int]
    )
  )

Pretty straight forward, we add or remove likes depending on if it’s a like or unlike and for comments we add or subtract 1 to the counter.

Idempotence​

But there is one issue with the comment counter, it’s not idempotent. Imagine a situation where the user click Publish comment twice (like my mother because she thinks it’s a file browser) then 2 events might get triggered. In that case we will wrongly count 2 instead of 1 comment.

If the API is well done the id should be generated by the frontend and therefore the duplicate event will have the same id. A solution would then be to use a Set[Id[Comment]] instead of the Int:

val commentCountByKey = commentsStream
  .groupByKey
  .aggregate(
    () => Set.empty[Id[Comment]],
    (_, comment: Comment, commentIds: Set[Id[Comment]]) =>
      if (comment.deleted) commentIds - comment.id else commentIds + comment.id,
    Materialized.`with`[Id[Post], Set[Id[Comment]], KeyValueStore[Bytes, Array[Byte]]](
      CirceSerdes.serde[Id[Post]],
      CirceSerdes.serde[Set[Id[Comment]]]
    )
  )
  .mapValues(_.size)

The Set would indeed guaranty the idempotence.

Semilattice

If you combine all Commutativity, Associativity and Idempotence you will obtain what we call a Semilattice. This is a concept to bear in mind when we design robust data pipeline.

It guarantees that not matter in which order the events are coming, how we apply our operations and if there is duplicate events, the result will always be the same.

Joins

We’ve started with KStreams, from them we built up a state with KTables, now it’s time to join all of them to create the long awaited DenormalisedPosts:

Can we improve something here?
We can always improve something… In this case the .to(“denormalised-posts”) doesn’t look very safe for exactly the same reasons we saw with the producers and the stream builder. Even though here it’s not to verbose, there is no guaranty at run time that the given topic is the one that contains the right type of document.

Again we can decorate the KStream type with a toTopic function that will reuse our Record type and give the right topic for the output type of the stream:

implicit class KStreamOps[K, V](stream: KStreamS[K, V]) {
  def toTopic(implicit record: Record[K, V], produced: Produced[K, V]) = stream.to(record.topic)
}

The usage becomes simply:

...
  .toStream
  .toTopic

Any other stuff we can improve?

Monocle

...
  .join(likesByKey,
        (denormalisedPost: DenormalisedPost, likes: Set[Like]) =>
          denormalisedPost.copy(interactions = denormalisedPost.interactions.copy(likes = likes)))
  .join(commentCountByKey,
        (denormalisedPost: DenormalisedPost, commentCount: Int) =>
          denormalisedPost.copy(interactions = denormalisedPost.interactions.copy(comments = commentCount)))

...
  .join(likesByKey,
        (denormalisedPost: DenormalisedPost, likes: Set[Like]) =>
          denormalisedPost.lens(_.interactions.likes).set(likes))
  .join(commentCountByKey,
        (denormalisedPost: DenormalisedPost, commentCount: Int) =>
          denormalisedPost.lens(_.interactions.comments).set(commentCount))