Kafka and the Micronaut Framework - Event-Driven Applications

Use Kafka to communicate between your Micronaut applications.

Authors: Burt Beckwith

Micronaut Version: 3.2.7

1. Getting Started

In this guide, we will create a Micronaut application written in Kotlin.

In this guide, we will create two microservices that will use Kafka to communicate with each other in an asynchronous and decoupled way.

2. What you will need

To complete this guide, you will need the following:

  • Some time on your hands

  • A decent text editor or IDE

  • JDK 1.8 or greater installed with JAVA_HOME configured appropriately

  • Docker and Docker Compose installed if you will be running Kafka in Docker, and for running tests.

3. Solution

We recommend that you follow the instructions in the next sections and create the application step by step. However, you can go right to the completed example.

4. Writing the application

Let’s describe the microservices you will build through the guide.

  • books - It returns a list of books. It uses a domain consisting of a book name and ISBN. It also publishes a message in Kafka every time a book is accessed.

  • analytics - It connects to Kafka to update the analytics for every book (a counter). It also exposes an endpoint to get the analytics.

4.1. Enable annotation Processing

If you use Java or Kotlin and IntelliJ IDEA, make sure to enable annotation processing.

annotationprocessorsintellij

4.2. Books Microservice

Create the books microservice using the Micronaut Command Line Interface or with Micronaut Launch.

mn create-app --features=kafka,reactor,graalvm,testcontainers example.micronaut.books --build=maven --lang=kotlin
If you don’t specify the --build argument, Gradle is used as the build tool.
If you don’t specify the --lang argument, Java is used as the language.

If you use Micronaut Launch, select Micronaut Application as application type and add the kafka, reactor, graalvm, and testcontainers features.

The previous command creates a directory named books and a Micronaut application inside it with default package example.micronaut.

In addition to the dependencies added by the testcontainers feature, we also need a test dependency for Kafka in Testcontainers, along with one for the Awaitility library:

pom.xml
<dependency>
    <groupId>org.testcontainers</groupId>
    <artifactId>kafka</artifactId>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.awaitility</groupId>
    <artifactId>awaitility</artifactId>
    <version>4.1.1</version>
    <scope>test</scope>
</dependency>

Create a Book POJO:

books/src/main/kotlin/example/micronaut/Book.kt
package example.micronaut

import io.micronaut.core.annotation.Introspected

@Introspected
data class Book(val isbn: String, val name: String)

To keep this guide simple there is no database persistence - BookService keeps the list of books in memory:

books/src/main/kotlin/example/micronaut/BookService.kt
package example.micronaut

import java.util.Optional
import javax.annotation.PostConstruct
import jakarta.inject.Singleton

@Singleton
class BookService {

    private val bookStore: MutableList<Book> = mutableListOf()

    @PostConstruct
    fun init() {
        bookStore.add(Book("1491950358", "Building Microservices"))
        bookStore.add(Book("1680502395", "Release It!"))
        bookStore.add(Book("0321601912", "Continuous Delivery"))
    }

    fun listAll(): List<Book> = bookStore

    fun findByIsbn(isbn: String): Optional<Book> =
            bookStore.stream()
                    .filter { (i) -> i == isbn }
                    .findFirst()
}

Create a BookController class to handle incoming HTTP requests to the books microservice:

books/src/main/kotlin/example/micronaut/BookController.kt
package example.micronaut

import io.micronaut.http.annotation.Controller
import io.micronaut.http.annotation.Get
import java.util.Optional

@Controller("/books") (1)
class BookController(private val bookService: BookService) { (2)

    @Get (3)
    fun listAll(): List<Book> = bookService.listAll()

    @Get("/{isbn}") (4)
    fun findBook(isbn: String): Optional<Book> = bookService.findByIsbn(isbn)
}
1 The @Controller annotation defines the class as a controller mapped to the root URI /books
2 Use constructor injection to inject a bean of type BookService
3 The @Get annotation maps the listAll method to an HTTP GET request on /books
4 The @Get annotation maps the findBook method to an HTTP GET request on /books/{isbn}

4.3. Analytics Microservice

Create the analytics microservice using the Micronaut Command Line Interface or with Micronaut Launch.

mn create-app --features=kafka,graalvm example.micronaut.analytics --build=maven --lang=kotlin
If you don’t specify the --build argument, Gradle is used as the build tool.
If you don’t specify the --lang argument, Java is used as the language.

If you use Micronaut Launch, select Micronaut Application as application type and add the kafka and graalvm features.

Create a Book POJO:

analytics/src/main/kotlin/example/micronaut/Book.kt
package example.micronaut

import io.micronaut.core.annotation.Introspected

@Introspected
data class Book(val isbn: String, val name: String)
This Book POJO is the same as the one in the books microservice. In a real application this would be in a shared library but to keep things simple we’ll just duplicate it.

Create a BookAnalytics POJO:

analytics/src/main/kotlin/example/micronaut/BookAnalytics.kt
package example.micronaut

import io.micronaut.core.annotation.Introspected

@Introspected
data class BookAnalytics(val bookIsbn: String, val count: Long)

To keep this guide simple there is no database persistence - AnalyticsService keeps book analytics in memory:

analytics/src/main/kotlin/example/micronaut/AnalyticsService.kt
package example.micronaut

import java.util.concurrent.ConcurrentHashMap
import jakarta.inject.Singleton

@Singleton
class AnalyticsService {

    private val bookAnalytics: MutableMap<Book, Long> = ConcurrentHashMap() (1)

    fun updateBookAnalytics(book: Book) { (2)
        bookAnalytics.compute(book) { k, v ->
            if (v == null) return@compute 1L else return@compute v + 1
        }
    }

    fun listAnalytics(): List<BookAnalytics> = (3)
            bookAnalytics.entries.map { (key, value) -> BookAnalytics(key.isbn, value) }
}
1 Keep the book analytics in memory
2 Initialize and update the analytics for the book passed as parameter
3 Return all the analytics

Write a test for AnalyticsService:

analytics/src/test/kotlin/example/micronaut/AnalyticsServiceTest.kt
package example.micronaut

import io.micronaut.test.extensions.junit5.annotation.MicronautTest
import org.junit.jupiter.api.Assertions.assertEquals
import org.junit.jupiter.api.Test
import jakarta.inject.Inject

@MicronautTest
class AnalyticsServiceTest {

    @Inject
    lateinit var analyticsService: AnalyticsService

    @Test
    fun testUpdateBookAnalyticsAndGetAnalytics() {
        val b1 = Book("1491950358", "Building Microservices")
        val b2 = Book("1680502395", "Release It!")

        analyticsService.updateBookAnalytics(b1)
        analyticsService.updateBookAnalytics(b1)
        analyticsService.updateBookAnalytics(b1)
        analyticsService.updateBookAnalytics(b2)

        val analytics = analyticsService.listAnalytics()

        assertEquals(2, analytics.size)
        assertEquals(3, findBookAnalytics(b1, analytics).count)
        assertEquals(1, findBookAnalytics(b2, analytics).count)
    }

    private fun findBookAnalytics(b: Book, analytics: List<BookAnalytics>): BookAnalytics {
        val ba : BookAnalytics? = analytics.filter { (bookIsbn) -> bookIsbn == b.isbn }.firstOrNull()
        return ba ?: throw RuntimeException("Book not found")
    }
}

Create a Controller to expose the analytics:

analytics/src/main/kotlin/example/micronaut/AnalyticsController.kt
package example.micronaut

import io.micronaut.http.annotation.Controller
import io.micronaut.http.annotation.Get

@Controller("/analytics")
class AnalyticsController(private val analyticsService: AnalyticsService) {

    @Get
    fun listAnalytics(): List<BookAnalytics> = analyticsService.listAnalytics() (1)
}
1 Just expose the analytics

The application doesn’t expose the method updateBookAnalytics created in AnalyticsService. This method will be invoked when reading messages from Kafka.

To run the tests:

analytics
./mvnw test

Modify the Application class to use dev as a default environment:

The Micronaut framework supports the concept of one or many default environments. A default environment is one that is only applied if no other environments are explicitly specified or deduced.

analytics/src/main/kotlin/example/micronaut/Application.kt
package example.micronaut

import io.micronaut.context.env.Environment.DEVELOPMENT
import io.micronaut.runtime.Micronaut.build

fun main(args: Array<String>) {
    build()
        .args(*args)
        .packages("example.micronaut")
        .defaultEnvironments(DEVELOPMENT)
        .start()
}

Create src/main/resources/application-dev.yml. The Micronaut framework applies this configuration file only for the dev environment.

analytics/src/main/resources/application-dev.yml
micronaut:
  server:
    port: 8081 (1)
1 Start the analytics microservice on port 8081

5. Running the application

Start the books microservice:

books
./mvnw mn:run
16:35:55.614 [main] INFO  io.micronaut.runtime.Micronaut - Startup completed in 576ms. Server Running: http://localhost:8080

Start the analytics microservice:

analytics
./mvnw mn:run
16:35:55.614 [main] INFO  io.micronaut.runtime.Micronaut - Startup completed in 623ms. Server Running: http://localhost:8081

You can use curl to test the application:

curl http://localhost:8080/books
[{"isbn":"1491950358","name":"Building Microservices"},{"isbn":"1680502395","name":"Release It!"},{"isbn":"0321601912","name":"Continuous Delivery"}]
curl http://localhost:8080/books/1491950358
{"isbn":"1491950358","name":"Building Microservices"}
curl http://localhost:8081/analytics
[]

Note that getting the analytics returns an empty list because the applications are not communicating with each other (yet).

6. Kafka and the Micronaut Framework

6.1. Install Kafka

A fast way to start using Kafka is via Docker. Create this docker-compose.yml file:

docker/docker-compose.yml
version: '2'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper
    ports:
      - 2181:2181 (1)
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
  kafka:
    image: confluentinc/cp-kafka
    depends_on:
      - zookeeper
    ports:
      - 9092:9092 (2)
    environment:
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
1 Zookeeper uses port 2181 by default, but you can change the value if necessary.
2 Kafka uses port 9092 by default, but you can change the value if necessary.

Start Zookeeper and Kafka (use CTRL-C to stop both):

docker-compose up

6.2. Books Microservice

The generated code includes configuration to connect to a Kafka broker running on localhost:9092. In case you want to change the configuration, update the following:

books/src/main/resources/application.yml
kafka:
  bootstrap:
    servers: localhost:9092

6.2.1. Create Kafka client (producer)

Let’s create an interface to send messages to Kafka. The Micronaut framework will implement the interface at compilation time:

books/src/main/kotlin/example/micronaut/AnalyticsClient.kt
package example.micronaut

import io.micronaut.configuration.kafka.annotation.KafkaClient
import io.micronaut.configuration.kafka.annotation.Topic
import org.reactivestreams.Publisher
import reactor.core.publisher.Mono

@KafkaClient
interface AnalyticsClient {

    @Topic("analytics") (1)
    fun updateAnalytics(book: Book) : Mono<Book> (2)
}
1 Set the topic name
2 Send the Book POJO. The Framework will automatically convert it to JSON before sending it

6.2.2. Create Tests

We could use mocks to test the message sending logic between BookController, AnalyticsFilter, and AnalyticsClient, but it’s more realistic to use a running Kafka broker. To avoid the burden of having to install Kafka locally (and to make the tests more CI-friendly) we’ll use Testcontainers to run Kafka inside a Docker container.

Write a test for BookController to verify the interaction with AnalyticsService:

books/src/test/kotlin/example/micronaut/BookControllerTest.kt
package example.micronaut

import io.micronaut.configuration.kafka.annotation.KafkaListener
import io.micronaut.configuration.kafka.annotation.OffsetReset.EARLIEST
import io.micronaut.configuration.kafka.annotation.Topic
import io.micronaut.core.type.Argument
import io.micronaut.http.HttpRequest
import io.micronaut.http.client.HttpClient
import io.micronaut.http.client.annotation.Client
import io.micronaut.http.client.exceptions.HttpClientResponseException
import io.micronaut.test.extensions.junit5.annotation.MicronautTest
import io.micronaut.test.support.TestPropertyProvider
import org.awaitility.Awaitility.await
import org.junit.jupiter.api.AfterEach
import org.junit.jupiter.api.Assertions.assertEquals
import org.junit.jupiter.api.Assertions.assertNotNull
import org.junit.jupiter.api.Assertions.assertThrows
import org.junit.jupiter.api.Assertions.assertTrue
import org.junit.jupiter.api.Test
import org.junit.jupiter.api.TestInstance
import org.junit.jupiter.api.TestInstance.Lifecycle.PER_CLASS
import org.testcontainers.containers.KafkaContainer
import org.testcontainers.junit.jupiter.Container
import org.testcontainers.junit.jupiter.Testcontainers
import org.testcontainers.utility.DockerImageName
import java.util.Optional
import java.util.concurrent.ConcurrentLinkedDeque
import java.util.concurrent.TimeUnit.SECONDS
import jakarta.inject.Inject

@Testcontainers (1)
@MicronautTest
@TestInstance(PER_CLASS) (2)
class BookControllerTest : TestPropertyProvider { (3)

    companion object {
        val received: MutableCollection<Book> = ConcurrentLinkedDeque()
    }

    @Container
    val kafka = KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:latest")) (4)

    @Inject
    lateinit var analyticsListener: AnalyticsListener (5)

    @Inject
    @field:Client("/")
    lateinit var client: HttpClient (6)

    @Test
    fun testMessageIsPublishedToKafkaWhenBookFound() {
        val isbn = "1491950358"

        val result : Optional<Book> = retrieveGet("/books/" + isbn) as Optional<Book> (7)
        assertNotNull(result)
        assertTrue(result.isPresent)
        assertEquals(isbn, result.get().isbn)

        await().atMost(5, SECONDS).until { !received.isEmpty() } (8)
        assertEquals(1, received.size) (9)

        val bookFromKafka = received.iterator().next()
        assertNotNull(bookFromKafka)
        assertEquals(isbn, bookFromKafka.isbn)
    }

    @Test
    fun testMessageIsNotPublishedToKafkaWhenBookNotFound() {
        assertThrows(HttpClientResponseException::class.java) { retrieveGet("/books/INVALID") }

        Thread.sleep(5_000); (10)
        assertEquals(0, received.size);
    }

    override fun getProperties(): Map<String, String> {
        kafka.start()
        return mapOf("kafka.bootstrap.servers" to kafka.bootstrapServers) (11)
    }

    @AfterEach
    fun cleanup() {
        received.clear()
    }

    private fun retrieveGet(url: String) = client
            .toBlocking()
            .retrieve(HttpRequest.GET<Any>(url),
                    Argument.of(Optional::class.java, Book::class.java))

    @KafkaListener(offsetReset = EARLIEST)
    class AnalyticsListener {

        @Topic("analytics")
        fun updateAnalytics(book: Book) {
            received.add(book)
        }
    }
}
1 Use the @Testcontainers annotation to configure automatic container management (not necessary in Spock tests)
2 Classes that implement TestPropertyProvider must use this annotation to create a single class instance for all tests (not necessary in Spock tests)
3 Implementing TestPropertyProvider allows the test class to provide Micronaut configuration properties, in this case the dynamically allocated Kafka broker port
4 The Testcontainer instance for Kafka
5 Dependency injection for the AnalyticsListener class declared below, a Kafka listener class that replicates the functionality of the class of the same name in the analytics microservice
6 Dependency injection for an HTTP client that the Micronaut framework will implement at compile to make calls to BookController
7 Use the HttpClient to retrieve a Book, which will trigger sending a message with Kafka
8 Wait a few seconds for the message to arrive; it should happen very quickly, but the message will be sent on a separate thread
9 Verify that the message was received and has the correct data
10 Wait a few seconds to make sure no message is sent
11 Configure the Kafka broker port (it will be different unused port each time) so Micronaut Kafka clients and listeners connect to the test broker

6.2.3. Send Analytics information automatically

Sending a message to Kafka is as simple as injecting AnalyticsClient and calling the updateAnalytics method. The goal is to do it automatically every time a book is returned, i.e., every time there is a call to http://localhost:8080/books/{isbn}. To achieve this we will create an Http Server Filter. Create the AnalyticsFilter class:

books/src/main/kotlin/example/micronaut/AnalyticsFilter.kt
package example.micronaut

import io.micronaut.http.HttpRequest
import io.micronaut.http.MutableHttpResponse
import io.micronaut.http.annotation.Filter
import io.micronaut.http.filter.HttpServerFilter
import io.micronaut.http.filter.ServerFilterChain
import reactor.core.publisher.Flux
import org.reactivestreams.Publisher

@Filter("/books/?*") (1)
class AnalyticsFilter(private val analyticsClient: AnalyticsClient) (3)
    : HttpServerFilter { (2)

    override fun doFilter(request: HttpRequest<*>, (4)
                          chain: ServerFilterChain): Publisher<MutableHttpResponse<*>> =
        Flux
            .from(chain.proceed(request)) (5)
            .flatMap { response: MutableHttpResponse<*> ->
                val book = response.getBody(Book::class.java).orElse(null) (6)
                if (book == null) {
                    Flux.just(response)
                }
                else {
                    Flux.from(analyticsClient.updateAnalytics(book)).map { b -> response } (7)
                }
            }
}
1 Annotate the class with @Filter and define the Ant-style matcher pattern to intercept all calls to the desired URIs
2 The class must implement HttpServerFilter
3 Dependency injection for the Kafka AnalyticsClient
4 Implement the doFilter method
5 Execute the request; this will invoke the controller action
6 Get the response from the controller and return the body as a Book
7 If the book is found, use the Kafka client to send a message

6.3. Analytics Microservice

6.3.1. Create Kafka consumer

Create a new class to act as a consumer of the messages sent to Kafka by the books microservice. The Micronaut framework will implement logic to invoke the consumer at compile time. Create the AnalyticsListener class:

analytics/src/main/kotlin/example/micronaut/AnalyticsListener.kt
package example.micronaut

import io.micronaut.configuration.kafka.annotation.KafkaListener
import io.micronaut.configuration.kafka.annotation.Topic
import io.micronaut.context.annotation.Requires
import io.micronaut.context.env.Environment

@Requires(notEnv = [Environment.TEST]) (1)
@KafkaListener (2)
class AnalyticsListener(private val analyticsService: AnalyticsService) { (3)

    @Topic("analytics") (4)
    fun updateAnalytics(book: Book) = analyticsService.updateBookAnalytics(book) (5)
}
1 Do not load this bean for the test environment - this lets us run the tests without having Kafka running
2 Annotate the class with @KafkaListener to indicate that this bean will consume messages from Kafka
3 Constructor injection for AnalyticsService
4 Annotate the method with @Topic and specify the topic name to use
5 Call AnalyticsService to update the analytics for the book

6.4. Running the application

Start the books microservice:

books
./mvnw mn:run
16:35:55.614 [main] INFO  io.micronaut.runtime.Micronaut - Startup completed in 576ms. Server Running: http://localhost:8080

Execute a curl request to get one book:

curl http://localhost:8080/books/1491950358
{"isbn":"1491950358","name":"Building Microservices"}

Start the analytics microservice:

analytics
./mvnw mn:run
16:35:55.614 [main] INFO  io.micronaut.runtime.Micronaut - Startup completed in 623ms. Server Running: http://localhost:8081

The application will consume and process the message automatically after startup.

Now, use curl to see the analytics:

curl http://localhost:8081/analytics
[{"bookIsbn":"1491950358","count":1}]

Update the curl command to the books microservice to retrieve other books and repeat the invocations, then re-run the curl command to the analytics microservice to see that the counts increase.

7. Generate Micronaut Application Native Images with GraalVM

We will use GraalVM, the polyglot embeddable virtual machine, to generate Native images of our Micronaut applications.

Compiling native images ahead-of-time with GraalVM improves startup time and reduces the memory footprint of JVM-based applications.

Only Java and Kotlin projects support using GraalVM’s native-image tool. Groovy relies heavily on reflection, which is only partially supported by GraalVM.

7.1. Native Image generation

The easiest way to install GraalVM on Linux or Mac is to use SDKMan.io.

Java 11
$ sdk install java 22.0.0.2.r11-grl
If you still use Java 8, use the JDK11 version of GraalVM.
Java 17
$ sdk install java 22.0.0.2.r17-grl

For installation on Windows, or for manual installation on Linux or Mac, see the GraalVM Getting Started documentation.

After installing GraalVM, install the native-image component, which is not installed by default:

gu install native-image

To generate native images for each application using Maven run:

./mvnw package -Dpackaging=native-image

The native image is created in the target directory and can be run with target/application.

Start the native images for the two microservices and run the same curl request as before to check that everything works with GraalVM.

8. Next steps

Read more about Kafka support in Micronaut framework.