SCALA-330: introduction to Kafka with Scala (#776)

* SCALA-330: introduction to kafka with scala (simple produce-consume flow)

* SCALA-330: introduction to kafka with scala (refactor produce-consume flow)

* SCALA-330: introduction to kafka with scala (add avro value serde)

* SCALA-330: introduction to kafka with scala (add transaction and isolate serde config)

* SCALA-330: introduction to kafka with scala (fix jsr310 date serialization)

* SCALA-330: introduction to kafka with scala (pin image version)

Author: exaucae

build.sbt

@@ -230,7 +230,6 @@ val fs2Version = "3.7.0"
 val AkkaVersion = "2.8.0"
 val AkkaHttpVersion = "10.5.0"
 val reactiveMongo = "1.0.10"
-val spireVersion = "0.18.0"
 
 lazy val scala_libraries = (project in file("scala-libraries"))
   .settings(
@@ -386,15 +385,34 @@ lazy val scala_libraries_4 = (project in file("scala-libraries-4"))
     IntegrationTest / fork := true
   )
 
+val spireVersion = "0.18.0"
+val kafkaVersion = "3.5.0"
+val pureconfigVersion = "0.17.4"
+val jackSonVersion = "2.15.1"
+val log4jApiScalaVersion = "12.0"
+val log4jVersion = "2.20.0"
+val avro4sVersion = "3.1.1"
+val kafkaAvroSerializer = "6.0.0"
+
 lazy val scala_libraries_5 = (project in file("scala-libraries-5"))
   .settings(
     name := "scala-libraries-5",
+    resolvers += "Kafka avro serializer" at "https://packages.confluent.io/maven",
     scalaVersion := scalaV,
     libraryDependencies ++= scalaTestDeps,
     libraryDependencies ++= Seq(
       sparkSqlDep,
       sparkCoreDep,
-      "org.typelevel" %% "spire" % spireVersion
+      "org.typelevel" %% "spire" % spireVersion,
+      "org.apache.kafka" % "kafka-clients" % kafkaVersion,
+      "com.github.pureconfig" %% "pureconfig" % pureconfigVersion,
+      "com.fasterxml.jackson.core" % "jackson-databind" % jackSonVersion,
+      "com.fasterxml.jackson.datatype" % "jackson-datatype-jsr310" % jackSonVersion,
+      "com.fasterxml.jackson.module" %% "jackson-module-scala" % jackSonVersion,
+      "com.sksamuel.avro4s" %% "avro4s-core" % avro4sVersion,
+      "io.confluent" % "kafka-avro-serializer" % kafkaAvroSerializer,
+      "org.apache.logging.log4j" %% "log4j-api-scala" % log4jApiScalaVersion,
+      "org.apache.logging.log4j" % "log4j-core" % log4jVersion % Runtime
     )
   )
 

scala-libraries-5/src/main/resources/docker-compose.yml

@@ -0,0 +1,58 @@
+version: "2"
+
+volumes:
+  kafka_data:
+    driver: local
+
+networks:
+  kafka_net:
+    driver: bridge
+  traccar-net:
+    driver: bridge
+
+services:
+  kafka:
+    image: docker.io/bitnami/kafka:3.4.0-debian-11-r23
+    networks:
+      - kafka_net
+    ports:
+      - "9094:9094"
+    volumes:
+      - "kafka_data:/bitnami"
+    environment:
+      - BITNAMI_DEBUG=yes
+      - ALLOW_PLAINTEXT_LISTENER=yes
+      - KAFKA_KRAFT_CLUSTER_ID=9YoavaRpTCOitT3Dm2OQFQ
+      # For more details see See https://rmoff.net/2018/08/02/kafka-listeners-explained/
+      - KAFKA_CFG_LISTENERS=CLIENT://:9092,CONTROLLER://:9093,EXTERNAL://:9094
+      - KAFKA_CFG_ADVERTISED_LISTENERS=CLIENT://kafka:9092,EXTERNAL://localhost:9094
+      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CLIENT:PLAINTEXT,CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT
+      - KAFKA_CFG_INTER_BROKER_LISTENER_NAME=CLIENT
+  kafka-schema-registry:
+    image: bitnami/schema-registry:6.0
+    networks:
+      - kafka_net
+    ports:
+      - "8081:8081"
+    depends_on:
+      kafka:
+        condition: service_started
+    environment:
+      SCHEMA_REGISTRY_DEBUG: true
+      SCHEMA_REGISTRY_KAFKA_BROKERS: PLAINTEXT://kafka:9092
+  kafka-ui:
+    image: provectuslabs/kafka-ui:v0.7.0
+    networks:
+      - kafka_net
+    ports:
+      - "8088:8080"
+    depends_on:
+      kafka:
+        condition: service_started
+      kafka-schema-registry:
+        condition: service_started
+    environment:
+      KAFKA_CLUSTERS_0_NAME: baeldung
+      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:9092
+      KAFKA_CLUSTERS_0_SCHEMAREGISTRY: http://kafka-schema-registry:8081
+      DYNAMIC_CONFIG_ENABLED: 'true'

scala-libraries-5/src/main/resources/kafka-intro-avro.conf

@@ -0,0 +1,19 @@
+bootstrap-servers = "localhost:9094"
+schema-registry-url = "http://localhost:8081"
+topic="scala-articles-avro"
+
+producer {
+    client.id = baeldung-scala-kafka-producer
+    bootstrap.servers = ${bootstrap-servers}
+    transactional.id = "baeldung-scala-kafka-producer"
+}
+
+serde {
+    schema.registry.url = ${schema-registry-url}
+}
+
+
+consumer {
+     group.id = baeldung-scala-kafka-consumer
+     bootstrap.servers = ${bootstrap-servers}
+}

scala-libraries-5/src/main/resources/kafka-intro.conf

@@ -0,0 +1,13 @@
+bootstrap-servers = "localhost:9094"
+topic="scala-articles"
+
+producer {
+    client.id = baeldung-scala-kafka-producer
+    bootstrap.servers = ${bootstrap-servers}
+}
+
+
+consumer {
+     group.id = baeldung-scala-kafka-consumer
+     bootstrap.servers = ${bootstrap-servers}
+}

scala-libraries-5/src/main/resources/log4j2.xml

@@ -0,0 +1,13 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<Configuration status="WARN">
+    <Appenders>
+        <Console name="Console" target="SYSTEM_OUT">
+            <PatternLayout pattern="%d{HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n"/>
+        </Console>
+    </Appenders>
+    <Loggers>
+        <Root level="info">
+            <AppenderRef ref="Console"/>
+        </Root>
+    </Loggers>
+</Configuration>

scala-libraries-5/src/main/scala-2/com/baeldung/scala/kafka/intro/common/Article.scala

@@ -0,0 +1,13 @@
+package com.baeldung.scala.kafka.intro.common
+
+import java.time.LocalDate
+
+case class Article(
+  id: String,
+  title: String,
+  content: String,
+  created: LocalDate,
+  author: Author
+)
+
+case class Author(id: Int, name: String)

scala-libraries-5/src/main/scala-2/com/baeldung/scala/kafka/intro/common/ClientConfig.scala

@@ -0,0 +1,16 @@
+package com.baeldung.scala.kafka.intro.common
+
+import com.typesafe.config.Config
+
+import scala.jdk.CollectionConverters._
+trait ClientConfig {
+  implicit class configMapperOps(config: Config) {
+    def asJavaMap: java.util.Map[String, AnyRef] = config.toMap.asJava
+
+    def toMap: Map[String, AnyRef] = config
+      .entrySet()
+      .asScala
+      .map(pair => (pair.getKey, config.getAnyRef(pair.getKey)))
+      .toMap
+  }
+}

scala-libraries-5/src/main/scala-2/com/baeldung/scala/kafka/intro/common/SerdeConfig.scala

@@ -0,0 +1,18 @@
+package com.baeldung.scala.kafka.intro.common
+
+import com.typesafe.config.Config
+import pureconfig.ConfigSource
+import pureconfig.generic.auto.exportReader
+
+import java.util
+
+case class SerdeConfig(serde: Config)
+
+object SerdeConfig extends ClientConfig {
+  def getConfig(resource: String): util.Map[String, AnyRef] = {
+    val source =
+      ConfigSource.resources(resource).loadOrThrow[SerdeConfig]
+    val serde = source.serde.asJavaMap
+    serde
+  }
+}

scala-libraries-5/src/main/scala-2/com/baeldung/scala/kafka/intro/consumer/ArticleAvroConsumer.scala

@@ -0,0 +1,58 @@
+package com.baeldung.scala.kafka.intro.consumer
+
+import com.baeldung.scala.kafka.intro.common.{Article, SerdeConfig}
+import com.baeldung.scala.kafka.intro.consumer.common.{
+  AvroDeSerializer,
+  ConsumerConfig,
+  ConsumerUtils
+}
+import com.sksamuel.avro4s.RecordFormat
+import org.apache.kafka.clients.consumer.KafkaConsumer
+import org.apache.kafka.common.serialization.{Deserializer, StringDeserializer}
+
+import scala.concurrent.duration.{FiniteDuration, MILLISECONDS}
+import scala.jdk.CollectionConverters._
+import scala.jdk.javaapi.CollectionConverters.asJavaCollection
+import scala.util.Try
+
+object ArticleAvroConsumer
+  extends App
+  with ConsumerUtils[Article]
+  with AvroDeSerializer {
+
+  private val (config, topic) =
+    ConsumerConfig.getConfig("kafka-intro-avro.conf")
+  private val serde = SerdeConfig.getConfig("kafka-intro-avro.conf")
+
+  val keyDeSerializer: StringDeserializer = new StringDeserializer()
+
+  implicit lazy val Valueformat: RecordFormat[Article] = RecordFormat[Article]
+  val valueDeSerializer: Deserializer[Article] = deserializer[Article]
+  valueDeSerializer.configure(serde, false)
+
+  private val consumer =
+    new KafkaConsumer(config, keyDeSerializer, valueDeSerializer)
+
+  consumer.subscribe(asJavaCollection(List(topic)))
+
+  consumer.seekToBeginning(Nil.asJava)
+
+  Try {
+    while (true) {
+      val messages = pool(consumer, FiniteDuration(1, MILLISECONDS))
+
+      for ((_, article) <- messages) {
+        logger.info(
+          s"New article received. Title: ${article.title} .  Author: ${article.author.name}, Date: ${article.created}  "
+        )
+      }
+    }
+  }.recover { case error =>
+    logger.error(error)
+    logger.error(
+      "Something went wrong when seeking messsages from begining. Unsubscribing"
+    )
+    consumer.unsubscribe();
+  }
+  consumer.close()
+}

scala-libraries-5/src/main/scala-2/com/baeldung/scala/kafka/intro/consumer/ArticleJsonStringConsumer.scala

@@ -0,0 +1,38 @@
+package com.baeldung.scala.kafka.intro.consumer
+
+import com.baeldung.scala.kafka.intro.common.Article
+import com.baeldung.scala.kafka.intro.consumer.common.{
+  ConsumerConfig,
+  ConsumerUtils,
+  JsonStringDeSerializer
+}
+import org.apache.kafka.clients.consumer.KafkaConsumer
+
+import java.util.concurrent.TimeUnit.SECONDS
+import scala.concurrent.duration.FiniteDuration
+import scala.jdk.javaapi.CollectionConverters.asJavaCollection
+
+object ArticleJsonStringConsumer
+  extends App
+  with ConsumerUtils[Article]
+  with JsonStringDeSerializer[Article] {
+
+  private val (config, topic) = ConsumerConfig.getConfig("kafka-intro.conf")
+
+  private val consumer =
+    new KafkaConsumer(config, keyDeSerializer, valueDeSerializer)
+
+  consumer.subscribe(asJavaCollection(List(topic)))
+
+  while (true) {
+    val messages = pool(consumer, FiniteDuration(1, SECONDS))
+    for ((_, value) <- messages) {
+      val article = fromJsonString(value)
+      logger.info(
+        s"New article received. Title: ${article.title} .  Author: ${article.author.name} "
+      )
+    }
+    consumer.commitAsync()
+  }
+
+}