Add a simple producer/consumer Kafka example

Author: carlosedp

src/main/resources/application.properties

@@ -33,6 +33,12 @@ mp.messaging.incoming.articles-in.topic=articles
 mp.messaging.incoming.articles-in.value.deserializer=org.acme.kafka.ArticleDeserializer
 mp.messaging.outgoing.articles-out.connector=smallrye-kafka
 mp.messaging.outgoing.articles-out.topic=articles-processed
+# The following is the Kafka configuration for the Hello World Kafka Example
+quarkus.kafka.devservices.topic-partitions.hello-world=1
+mp.messaging.outgoing.hello-world-out.connector=smallrye-kafka
+mp.messaging.outgoing.hello-world-out.topic=hello-world
+mp.messaging.incoming.hello-world-in.connector=smallrye-kafka
+mp.messaging.incoming.hello-world-in.topic=hello-world
 ### -------------------PERSISTENCE CONFIGURATION------------------------- ###
 # Datasource configuration
 quarkus.datasource.db-kind=postgresql

src/main/scala/org/acme/kafka/helloworld/HelloWorldConsumer.scala

@@ -0,0 +1,18 @@
+package org.acme.kafka.helloworld
+
+import io.quarkus.logging.Log
+
+import io.smallrye.reactive.messaging.annotations.Blocking
+import jakarta.enterprise.context.ApplicationScoped
+import org.eclipse.microprofile.metrics.annotation.Counted
+import org.eclipse.microprofile.reactive.messaging.Incoming
+
+// This is a simple Kafka consumer that reads messages from the "hello-world-out" channel and logs them.
+
+@ApplicationScoped
+class HelloWorldConsumer:
+  @Incoming("hello-world-in")
+  @Blocking
+  @Counted(name = "consumedHelloMessages", description = "How many hello messages were consumed")
+  def consumeMessage(msg: String): Unit =
+    Log.info(s"Received message: ${msg}")

src/main/scala/org/acme/kafka/helloworld/HelloWorldProducer.scala

@@ -0,0 +1,30 @@
+package org.acme.kafka.helloworld
+
+import java.util.concurrent.TimeUnit
+
+import io.quarkus.logging.Log
+import io.quarkus.scheduler.Scheduled
+
+import jakarta.enterprise.context.ApplicationScoped
+import jakarta.transaction.Transactional
+import org.eclipse.microprofile.metrics.annotation.Counted
+import org.eclipse.microprofile.reactive.messaging.{Channel, Emitter}
+
+// This is a simple Kafka producer that sends a message every second to the Kafka topic "hello-world" via the channel "hello-world-in" with a sequential ID and an identifier from the PRODUCER_ID environment variable.
+
+@ApplicationScoped
+class HelloWorldProducer(
+    @Channel("hello-world-out") emitter: Emitter[String]
+  ):
+  // Store a sequential ID for the messages
+  private var id = 0
+  @Transactional
+  @Scheduled(every = "1s", delay = 5, delayUnit = TimeUnit.SECONDS, identity = "hello-world-producer")
+  @Counted(name = "producedHelloMessages", description = "How many hello messages were produced")
+  def produce() =
+    val producerId = scala.util.Properties.envOrElse("PRODUCER_ID", "unknown")
+    val message    = s"Hello World from $producerId - $id"
+    Log.info("Producing a message: " + message)
+    emitter.send(message)
+    id += 1
+end HelloWorldProducer