Apache Kafka is a distributed stream processing platform that can be used for a range of messaging requirements in addition to stream processing and real-time data handling.

Micronaut features dedicated support for defining both Kafka Producer and Consumer instances. Micronaut applications built with Kafka can be deployed with or without the presence of an HTTP server.

With Micronaut’s efficient compile-time AOP and cloud native features, writing efficient Kafka consumer applications that use very little resources is a breeze.

2 Release History

For this project, you can find a list of releases (with release notes) here:

https://github.com/micronaut-projects/micronaut-kafka/releases

Upgrading to Micronaut Kafka 5.0

Micronaut Kafka 5.0 is a significant major version which includes a number of changes you will need to consider when upgrading.

Micronaut 4, Kafka 3 & Java 17 baseline

Micronaut Kafka 5.0 requires the following minimum set of dependencies:

Java 17 or above
Kafka 3
Micronaut 4 or above

`@KafkaClient` no longer recoverable by default

Previous versions of Micronaut Kafka used the meta-annotation @Recoverable on the @KafkaClient annotation allowing you to define fallbacks in the case of failure. Micronaut Kafka 5 no longer includes this meta annotation and if you use fallbacks you should explicitly declare a dependency on io.micronaut:micronaut-retry and declare the @Recoverable explicitly.

Open Tracing No Longer Supported

Micronaut Kafka 5 no longer supports Open Tracing (which is deprecated and no longer maintained) and if you need distributed tracing you should instead use Open Telemetry.

3 Using the Micronaut CLI

To create a project with Kafka support using the Micronaut CLI, supply the kafka feature to the features flag.

$ mn create-app my-kafka-app --features kafka

This will create a project with the minimum necessary configuration for Kafka.

Kafka Messaging Application

The Micronaut CLI includes the ability to create Kafka-based messaging applications designed to implement message-driven microservices.

To create a Message-Driven Microservice with Micronaut + Kafka use the create-messaging-app command:

$ mn create-messaging-app my-kafka-app --features kafka

As you’d expect, you can start the application with ./gradlew run (for Gradle) or ./mvnw compile exec:exec (Maven). The application will (with the default config) attempt to connect to Kafka at http://localhost:9092, and will continue to run without starting up an HTTP server. All communication to/from the service will take place via Kafka producers and/or listeners.

Within the new project, you can now run the Kafka-specific code generation commands:

$ mn create-kafka-producer MessageProducer
| Rendered template Producer.java to destination src/main/java/my/kafka/app/MessageProducer.java

$ mn create-kafka-listener MessageListener
| Rendered template Listener.java to destination src/main/java/my/kafka/app/MessageListener.java

See the guide for Kafka and the Micronaut Framework - Event-driven Applications to learn more.

4 Kafka Quick Start

To add support for Kafka to an existing project, you should first add the Micronaut Kafka configuration to your build configuration. For example in Gradle:

implementation("io.micronaut.kafka:micronaut-kafka")

<dependency>
    <groupId>io.micronaut.kafka</groupId>
    <artifactId>micronaut-kafka</artifactId>
</dependency>

Configuring Kafka

The minimum requirement to configure Kafka is set the value of the kafka.bootstrap.servers property in application.yml:

Configuring Kafka

kafka.bootstrap.servers=localhost:9092

kafka:
    bootstrap:
        servers: localhost:9092

[kafka]
  [kafka.bootstrap]
    servers="localhost:9092"

kafka {
  bootstrap {
    servers = "localhost:9092"
  }
}

{
  kafka {
    bootstrap {
      servers = "localhost:9092"
    }
  }
}

{
  "kafka": {
    "bootstrap": {
      "servers": "localhost:9092"
    }
  }
}

The value can also be list of available servers:

Configuring Kafka

kafka.bootstrap.servers[0]=foo:9092
kafka.bootstrap.servers[1]=bar:9092

kafka:
    bootstrap:
        servers:
            - foo:9092
            - bar:9092

[kafka]
  [kafka.bootstrap]
    servers=[
      "foo:9092",
      "bar:9092"
    ]

kafka {
  bootstrap {
    servers = ["foo:9092", "bar:9092"]
  }
}

{
  kafka {
    bootstrap {
      servers = ["foo:9092", "bar:9092"]
    }
  }
}

{
  "kafka": {
    "bootstrap": {
      "servers": ["foo:9092", "bar:9092"]
    }
  }
}

You can also set the environment variable KAFKA_BOOTSTRAP_SERVERS to a comma separated list of values to externalize configuration.

If your broker needs to setup SSL, it can be configured this way:

Configuring Kafka

kafka.bootstrap.servers=localhost:9092
kafka.ssl.keystore.location=/path/to/client.keystore.p12
kafka.ssl.keystore.password=secret
kafka.ssl.truststore.location=/path/to/client.truststore.jks
kafka.ssl.truststore.password=secret
kafka.ssl.truststore.type=PKCS12
kafka.security.protocol=ssl

kafka:
  bootstrap:
    servers: localhost:9092
  ssl:
    keystore:
      location: /path/to/client.keystore.p12
      password: secret
    truststore:
      location: /path/to/client.truststore.jks
      password: secret
      type: PKCS12
  security:
    protocol: ssl

[kafka]
  [kafka.bootstrap]
    servers="localhost:9092"
  [kafka.ssl]
    [kafka.ssl.keystore]
      location="/path/to/client.keystore.p12"
      password="secret"
    [kafka.ssl.truststore]
      location="/path/to/client.truststore.jks"
      password="secret"
      type="PKCS12"
  [kafka.security]
    protocol="ssl"

kafka {
  bootstrap {
    servers = "localhost:9092"
  }
  ssl {
    keystore {
      location = "/path/to/client.keystore.p12"
      password = "secret"
    }
    truststore {
      location = "/path/to/client.truststore.jks"
      password = "secret"
      type = "PKCS12"
    }
  }
  security {
    protocol = "ssl"
  }
}

{
  kafka {
    bootstrap {
      servers = "localhost:9092"
    }
    ssl {
      keystore {
        location = "/path/to/client.keystore.p12"
        password = "secret"
      }
      truststore {
        location = "/path/to/client.truststore.jks"
        password = "secret"
        type = "PKCS12"
      }
    }
    security {
      protocol = "ssl"
    }
  }
}

{
  "kafka": {
    "bootstrap": {
      "servers": "localhost:9092"
    },
    "ssl": {
      "keystore": {
        "location": "/path/to/client.keystore.p12",
        "password": "secret"
      },
      "truststore": {
        "location": "/path/to/client.truststore.jks",
        "password": "secret",
        "type": "PKCS12"
      }
    },
    "security": {
      "protocol": "ssl"
    }
  }
}

Creating a Kafka Producer with @KafkaClient

To create a Kafka Producer that sends messages you can simply define an interface that is annotated with @KafkaClient.

For example the following is a trivial @KafkaClient interface:

ProductClient.java

import io.micronaut.configuration.kafka.annotation.KafkaClient;
import io.micronaut.configuration.kafka.annotation.KafkaKey;
import io.micronaut.configuration.kafka.annotation.Topic;
import io.micronaut.context.annotation.Requires;

@KafkaClient // (1)
public interface ProductClient {

    @Topic("my-products") // (2)
    void sendProduct(@KafkaKey String brand, String name); // (3)

    void sendProduct(@Topic String topic, @KafkaKey String brand, String name); // (4)
}

ProductClient.java

import io.micronaut.configuration.kafka.annotation.KafkaClient
import io.micronaut.configuration.kafka.annotation.KafkaKey
import io.micronaut.configuration.kafka.annotation.Topic
import io.micronaut.context.annotation.Requires

@KafkaClient // (1)
public interface ProductClient {

    @Topic('my-products') // (2)
    void sendProduct(@KafkaKey String brand, String name) // (3)

    void sendProduct(@Topic String topic, @KafkaKey String brand, String name) // (4)
}

ProductClient.java

import io.micronaut.configuration.kafka.annotation.KafkaClient
import io.micronaut.configuration.kafka.annotation.KafkaKey
import io.micronaut.configuration.kafka.annotation.Topic
import io.micronaut.context.annotation.Requires

@KafkaClient // (1)
interface ProductClient {
    @Topic("my-products")  // (2)
    fun sendProduct(@KafkaKey brand: String, name: String) // (3)

    fun sendProduct(@Topic topic: String, @KafkaKey brand: String, name: String) // (4)
}

1	The @KafkaClient annotation is used to designate this interface as a client
2	The @Topic annotation indicates which topics the `ProducerRecord` should be published to
3	The method defines two parameters: The parameter that is the Kafka key and the value.
4	It is also possible for the topic to be dynamic by making it a method argument

You can omit the key, however this will result in a null key which means Kafka will not know how to partition the record.

At compile time Micronaut will produce an implementation of the above interface. You can retrieve an instance of ProductClient either by looking up the bean from the ApplicationContext or by injecting the bean with @Inject:

Using ProductClient

ProductClient client = applicationContext.getBean(ProductClient.class);
client.sendProduct("Nike", "Blue Trainers");

Using ProductClient

val client = beanContext.getBean(ProductClient::class.java)
client.sendProduct("Nike", "Blue Trainers")

Note that since the sendProduct method returns void this means the method will send the ProducerRecord and block until the response is received. You can specify an executor and return either a CompletableFuture or Publisher to support non-blocking message delivery.

Creating a Kafka Consumer with @KafkaListener

To listen to Kafka messages you can use the @KafkaListener annotation to define a message listener.

The following example will listen for messages published by the ProductClient in the previous section:

ProductListener.java

import io.micronaut.configuration.kafka.annotation.KafkaKey;
import io.micronaut.configuration.kafka.annotation.KafkaListener;
import io.micronaut.configuration.kafka.annotation.OffsetReset;
import io.micronaut.configuration.kafka.annotation.Topic;
import io.micronaut.context.annotation.Requires;
import org.slf4j.Logger;

import static org.slf4j.LoggerFactory.getLogger;

@KafkaListener(offsetReset = OffsetReset.EARLIEST) // (1)
public class ProductListener {
    private static final Logger LOG = getLogger(ProductListener.class);

    @Topic("my-products") // (2)
    public void receive(@KafkaKey String brand, String name) { // (3)
        LOG.info("Got Product - {} by {}", name, brand);
    }
}

ProductListener.java

import io.micronaut.configuration.kafka.annotation.KafkaListener
import io.micronaut.configuration.kafka.annotation.OffsetReset
import io.micronaut.configuration.kafka.annotation.Topic
import io.micronaut.context.annotation.Requires

@Slf4j
@KafkaListener(offsetReset = OffsetReset.EARLIEST) // (1)
class ProductListener {

    @Topic('my-products') // (2)
    void receive(@KafkaKey String brand, String name) { // (3)
        log.info("Got Product - {} by {}", name, brand)
    }
}

ProductListener.java

import io.micronaut.configuration.kafka.annotation.KafkaKey
import io.micronaut.configuration.kafka.annotation.KafkaListener
import io.micronaut.configuration.kafka.annotation.OffsetReset
import io.micronaut.configuration.kafka.annotation.Topic
import io.micronaut.context.annotation.Requires
import org.slf4j.LoggerFactory

@KafkaListener(offsetReset = OffsetReset.EARLIEST) // (1)
class ProductListener {
    companion object {
        private val LOG = LoggerFactory.getLogger(ProductListener::class.java)
    }

    @Topic("my-products") // (2)
    fun receive(@KafkaKey brand: String, name: String) { // (3)
        LOG.info("Got Product - {} by {}", name, brand)
    }
}

1	The @KafkaListener is used with `offsetReset` set to `EARLIEST` which makes the listener start listening to messages from the beginning of the partition.
2	The @Topic annotation is again used to indicate which topic(s) to subscribe to.
3	The `receive` method defines 2 arguments: The argument that will receive the key and the argument that will receive the value.

Disabling Kafka

If for some reason, you need to disable the creation of kafka consumers, or producers, you can through configuration:

Disabling Kafka

kafka.enabled=false

kafka:
  enabled: false

[kafka]
  enabled=false

kafka {
  enabled = false
}

{
  kafka {
    enabled = false
  }
}

{
  "kafka": {
    "enabled": false
  }
}

5 Kafka Producers Using @KafkaClient

5.1 Defining @KafkaClient Methods

Specifying the Key and the Value

The Kafka key can be specified by providing a parameter annotated with @KafkaKey. If no such parameter is specified the record is sent with a null key.

The value to send is resolved by selecting the argument annotated with @MessageBody, otherwise the first argument with no specific binding annotation is used. For example:

    @Topic("my-products")
    void sendProduct(@KafkaKey String brand, String name);

    @Topic('my-products')
    void sendProduct(@KafkaKey String brand, String name)

    @Topic("my-products")
    fun sendProduct(@KafkaKey brand: String, name: String)

The method above will use the parameter brand as the key and the parameter name as the value.

Including Message Headers

There are a number of ways you can include message headers. One way is to annotate an argument with the @MessageHeader annotation and include a value when calling the method:

    @Topic("my-products")
    void sendProduct(@KafkaKey String brand, String name, @MessageHeader("My-Header") String myHeader);

    @Topic('my-products')
    void sendProduct(@KafkaKey String brand, String name, @MessageHeader('My-Header') String myHeader)

    @Topic("my-products")
    fun sendProduct(@KafkaKey brand: String, name: String, @MessageHeader("My-Header") myHeader: String)

The example above will include the value of the myHeader argument as a header called My-Header.

Another way to include headers is at the type level with the values driven from configuration:

Declaring @KafkaClient Headers

import io.micronaut.configuration.kafka.annotation.KafkaClient;
import io.micronaut.messaging.annotation.MessageHeader;

@KafkaClient(id="product-client")
@MessageHeader(name = "X-Token", value = "${my.application.token}")
public interface ProductClient {
    // define client API
}

Declaring @KafkaClient Headers

import io.micronaut.configuration.kafka.annotation.KafkaClient
import io.micronaut.messaging.annotation.MessageHeader

@KafkaClient(id='product-client')
@MessageHeader(name = 'X-Token', value = '${my.application.token}')
interface ProductClient {
    // define client API
}

Declaring @KafkaClient Headers

import io.micronaut.configuration.kafka.annotation.KafkaClient
import io.micronaut.messaging.annotation.MessageHeader

@KafkaClient(id = "product-client")
@MessageHeader(name = "X-Token", value = "\${my.application.token}")
interface ProductClient {
// define client API
}

The above example will send a header called X-Token with the value read from the setting my.application.token in application.yml (or the environnment variable MY_APPLICATION_TOKEN).

If the my.application.token is not set then an error will occur creating the client.

It is also possible to pass Collection<Header> or Headers object as method arguments as seen below.

Collection<Header> Argument

    @Topic("my-bicycles")
    void sendBicycle(@KafkaKey String brand, String model, Collection<Header> headers);

Collection<Header> Argument

    @Topic('my-bicycles')
    void sendBicycle(@KafkaKey String brand, String model, Collection<Header> headers)

Collection<Header> Argument

    @Topic("my-bicycles")
    fun sendBicycle(@KafkaKey brand: String, model: String, headers: Collection<Header>)

Kafka Header Javadocs

Headers Argument

    @Topic("my-bicycles")
    void sendBicycle(@KafkaKey String brand, String model, Headers headers);

Headers Argument

    @Topic('my-bicycles')
    void sendBicycle(@KafkaKey String brand, String model, Headers headers)

Headers Argument

    @Topic("my-bicycles")
    fun sendBicycle(@KafkaKey brand: String, model: String, headers: Headers)

Kafka Headers Javadocs

In the above examples, all of the key/value pairs in headers will be added to the list of headers produced to the topic. Header and Headers are part of the kafka-clients library:

Reactive and Non-Blocking Method Definitions

The @KafkaClient annotation supports the definition of reactive return types (such as Flowable or Reactor Flux) as well as Futures.

The KafkaProducer used internally to implement @KafkaClient support is inherently blocking, even though some of its methods describe themselves as "asynchronous". Configuring an executor (as shown in the following examples) is required in order to guarantee that a returned reactive type or Future will not block the calling thread.

Add the library Micronaut Reactor or Micronaut RxJava 3 to your application’s dependencies.

The following sections, which use Micronaut Reactor, cover advised configuration and possible method signatures and behaviour:

Configuring An Executor

As the send method of KafkaProducer can block the calling thread, it is recommended that you specify an executor to be used when returning either reactive types or CompletableFuture. This will ensure that the send logic is executed on a separate thread from that of the caller, and avoid undesirable conditions such as blocking of the Micronaut server’s event loop.

The executor to be used may be specified via configuration properties as in the following example:

kafka.producers.default.executor=blocking
kafka.producers.my-named-producer.executor=io

kafka:
    producers:
        default:
            executor: blocking
        my-named-producer:
            executor: io

[kafka]
  [kafka.producers]
    [kafka.producers.default]
      executor="blocking"
    [kafka.producers.my-named-producer]
      executor="io"

kafka {
  producers {
    'default' {
      executor = "blocking"
    }
    myNamedProducer {
      executor = "io"
    }
  }
}

{
  kafka {
    producers {
      default {
        executor = "blocking"
      }
      my-named-producer {
        executor = "io"
      }
    }
  }
}

{
  "kafka": {
    "producers": {
      "default": {
        "executor": "blocking"
      },
      "my-named-producer": {
        "executor": "io"
      }
    }
  }
}

Alternatively, the executor may be specified via the executor property of the @KafkaClient annotation:

@KafkaClient(value = "product-client", executor = TaskExecutors.BLOCKING)
public interface BookClient {

Note that an executor specified in the annotation will take precedent over that specified in the application configuration properties.

Mono Value and Return Type

    @Topic("my-books")
    Mono<Book> sendBook(@KafkaKey String author, Mono<Book> book);

    @Topic('my-books')
    Mono<Book> sendBook(@KafkaKey String author, Mono<Book> book);

    @Topic("my-books")
    fun sendBook(@KafkaKey author: String, book: Mono<Book>): Mono<Book>

The implementation will return a Mono that when subscribed to will subscribe to the passed Mono and send the emitted item as a ProducerRecord emitting the item again if successful or an error otherwise.

Flux Value and Return Type

    @Topic("my-books")
    Flux<RecordMetadata> sendBooks(@KafkaKey String author, Flux<Book> book);

    @Topic('my-books')
    Flux<RecordMetadata> sendBooks(@KafkaKey String author, Flux<Book> book);

    @Topic("my-books")
    fun sendBooks(@KafkaKey author: String, book: Flux<Book>): Flux<RecordMetadata>

The implementation will return a Reactor Flux that when subscribed to will subscribe to the passed Flux and for each emitted item will send a ProducerRecord emitting the resulting Kafka RecordMetadata if successful or an error otherwise.

Available Annotations

There are a number of annotations available that allow you to specify how a method argument is treated.

The following table summarizes the annotations and their purpose, with an example:

Table 1. Kafka Messaging Annotations
Annotation	Description	Example
@MessageBody	Allows explicitly indicating the body of the message to sent	`@MessageBody Product product`
@MessageHeader	Allows specifying a parameter that should be sent as a header	`@MessageHeader("X-My-Header") String myHeader`
@KafkaKey	Allows specifying the parameter that is the Kafka key	`@KafkaKey String key`
@KafkaPartition	Allows specifying the parameter that is the partition number	`@KafkaPartition Integer partition`
@KafkaPartitionKey	Allows specifying the parameter that is used to compute a partition number independently from the Message Key.	`@KafkaPartition String partitionKey`

For example, you can use the @MessageHeader annotation to bind a parameter value to a header in the ProducerRecord.

5.2 Configuring @KafkaClient beans

@KafkaClient and Producer Properties

There are a number of ways to pass configuration properties to the KafkaProducer. You can set default producer properties using kafka.producers.default in application.yml:

Applying Default Configuration

kafka.producers.default.retries=5
kafka.producers.default.bootstrap.servers=localhost:9096

kafka:
    producers:
        default:
            retries: 5
            bootstrap:
              servers: localhost:9096

[kafka]
  [kafka.producers]
    [kafka.producers.default]
      retries=5
      [kafka.producers.default.bootstrap]
        servers="localhost:9096"

kafka {
  producers {
    'default' {
      retries = 5
      bootstrap {
        servers = "localhost:9096"
      }
    }
  }
}

{
  kafka {
    producers {
      default {
        retries = 5
        bootstrap {
          servers = "localhost:9096"
        }
      }
    }
  }
}

{
  "kafka": {
    "producers": {
      "default": {
        "retries": 5,
        "bootstrap": {
          "servers": "localhost:9096"
        }
      }
    }
  }
}

Any property in the ProducerConfig class can be set, including any overrides over the global Micronaut Kafka configs. The above example will set the default number of times to retry sending a record as well as override kafka.bootstrap.servers.

Per @KafkaClient Producer Properties

To configure different properties for each client, you should set a @KafkaClient id using the annotation:

Using a Client ID

@KafkaClient("product-client")

Using a Client ID

@KafkaClient('product-client')

Using a Client ID

@KafkaClient("product-client")

This serves 2 purposes. Firstly it sets the value of the client.id setting used to build the Producer. Secondly, it allows you to apply per producer configuration in application.yml:

Applying Default Configuration

kafka.producers.product-client.retries=5
kafka.producers.product-client.bootstrap.servers=localhost:9097
kafka.producers.product-client-2.bootstrap.servers=localhost:9098

kafka:
    producers:
        product-client:
            retries: 5
            bootstrap:
              servers: localhost:9097
        product-client-2:
            bootstrap:
              servers: localhost:9098

[kafka]
  [kafka.producers]
    [kafka.producers.product-client]
      retries=5
      [kafka.producers.product-client.bootstrap]
        servers="localhost:9097"
    [kafka.producers.product-client-2]
      [kafka.producers.product-client-2.bootstrap]
        servers="localhost:9098"

kafka {
  producers {
    productClient {
      retries = 5
      bootstrap {
        servers = "localhost:9097"
      }
    }
    productClient2 {
      bootstrap {
        servers = "localhost:9098"
      }
    }
  }
}

{
  kafka {
    producers {
      product-client {
        retries = 5
        bootstrap {
          servers = "localhost:9097"
        }
      }
      product-client-2 {
        bootstrap {
          servers = "localhost:9098"
        }
      }
    }
  }
}

{
  "kafka": {
    "producers": {
      "product-client": {
        "retries": 5,
        "bootstrap": {
          "servers": "localhost:9097"
        }
      },
      "product-client-2": {
        "bootstrap": {
          "servers": "localhost:9098"
        }
      }
    }
  }
}

Finally, the @KafkaClient annotation itself provides a properties member that you can use to set producer specific properties:

Configuring Producer Properties with @KafkaClient

import io.micronaut.configuration.kafka.annotation.KafkaClient;
import io.micronaut.context.annotation.Property;
import io.micronaut.context.annotation.Requires;
import org.apache.kafka.clients.producer.ProducerConfig;

@KafkaClient(
    id = "product-client",
    acks = KafkaClient.Acknowledge.ALL,
    properties = @Property(name = ProducerConfig.RETRIES_CONFIG, value = "5")
)
public interface ProductClient {
    // define client API
}

Configuring Producer Properties with @KafkaClient

import io.micronaut.configuration.kafka.annotation.KafkaClient
import io.micronaut.context.annotation.Property
import io.micronaut.context.annotation.Requires
import org.apache.kafka.clients.producer.ProducerConfig

@KafkaClient(
        id = "product-client",
        acks = KafkaClient.Acknowledge.ALL,
        properties = @Property(name = ProducerConfig.RETRIES_CONFIG, value = '5')
)
interface ProductClient {
    // define client API
}

Configuring Producer Properties with @KafkaClient

import io.micronaut.configuration.kafka.annotation.KafkaClient
import io.micronaut.context.annotation.Property
import io.micronaut.context.annotation.Requires
import org.apache.kafka.clients.producer.ProducerConfig

@KafkaClient(
    id = "product-client",
    acks = KafkaClient.Acknowledge.ALL,
    properties = [Property(name = ProducerConfig.RETRIES_CONFIG, value = "5")]
)
interface ProductClient {
    // define client API
}

@KafkaClient and Serializers

When serializing keys and values Micronaut will by default attempt to automatically pick a Serializer to use. This is done via the CompositeSerdeRegistry bean.

You can replace the default SerdeRegistry bean with your own implementation by defining a bean that uses @Replaces(CompositeSerdeRegistry.class). See the section on Bean Replacement.

All common java.lang types (String, Integer, primitives etc.) are supported and for POJOs by default a Jackson based JSON serializer is used.

You can, however, explicitly override the Serializer used by providing the appropriate configuration in application.yml:

Applying Default Configuration

kafka.producers.product-client.value.serializer=org.apache.kafka.common.serialization.ByteArraySerializer

kafka:
    producers:
        product-client:
            value:
                serializer: org.apache.kafka.common.serialization.ByteArraySerializer

[kafka]
  [kafka.producers]
    [kafka.producers.product-client]
      [kafka.producers.product-client.value]
        serializer="org.apache.kafka.common.serialization.ByteArraySerializer"

kafka {
  producers {
    productClient {
      value {
        serializer = "org.apache.kafka.common.serialization.ByteArraySerializer"
      }
    }
  }
}

{
  kafka {
    producers {
      product-client {
        value {
          serializer = "org.apache.kafka.common.serialization.ByteArraySerializer"
        }
      }
    }
  }
}

{
  "kafka": {
    "producers": {
      "product-client": {
        "value": {
          "serializer": "org.apache.kafka.common.serialization.ByteArraySerializer"
        }
      }
    }
  }
}

You may want to do this if for example you choose an alternative serialization format such as Avro or Protobuf.

5.3 Sending Records in Batch

By default if you define a method that takes a container type such as a List the list will be serialized using the specified value.serializer (the default will result in a JSON array).

For example the following two methods will both send serialized arrays:

Sending Arrays and Lists

@Topic("books")
void sendList(List<Book> books);

@Topic("books")
void sendBooks(Book...books);

Sending Arrays and Lists

@Topic('books')
void sendList(List<Book> books)

@Topic('books')
void sendBooks(Book...books)

Sending Arrays and Lists

@Topic("books")
fun sendList(books: List<Book>)

@Topic("books")
fun sendBooks(vararg books: Book)

Instead of a sending a serialized array you may wish to instead send batches of ProducerRecord either synchronously or asynchronously.

To do this you can specify a value of true to the batch member of the @KafkaClient annotation:

Sending ProducerRecord batches

@KafkaClient(batch = true)
public interface BookClient {

    @Topic("books")
    void sendList(List<Book> books);

Sending ProducerRecord batches

@KafkaClient(batch = true)
interface BookClient {

    @Topic('books')
    void sendList(List<Book> books)

Sending ProducerRecord batches

@KafkaClient(batch = true)
interface BookClient {

    @Topic("books")
    fun sendList(books: List<Book>)

In the above case instead of sending a serialized array the client implementation will iterate over each item in the list and send a ProducerRecord for each. The previous example is blocking, however you can return a reactive type if desired:

Sending ProducerRecord batches Reactively

@KafkaClient(batch = true)
public interface BookClient {

    @Topic("books")
    Flux<RecordMetadata> send(List<Book> books);

Sending ProducerRecord batches Reactively

@KafkaClient(batch = true)
interface BookClient {

    @Topic('books')
    Flux<RecordMetadata> send(List<Book> books)

Sending ProducerRecord batches Reactively

@KafkaClient(batch = true)
interface BookClient {

    @Topic("books")
    fun send(books: List<Book>): Flux<RecordMetadata>

You can also use an unbound reactive type such as Flux as the source of your batch data:

Sending ProducerRecord batches from a Flux

@KafkaClient(batch = true)
public interface BookClient {

    @Topic("books")
    Flux<RecordMetadata> send(Flux<Book> books);

Sending ProducerRecord batches from a Flux

@KafkaClient(batch = true)
interface BookClient {

    @Topic('books')
    Flux<RecordMetadata> send(Flux<Book> books)

Sending ProducerRecord batches from a Flux

@KafkaClient(batch = true)
interface BookClient {

    @Topic("books")
    fun send(books: Flux<Book>): Flux<RecordMetadata>

5.4 Injecting Kafka Producer Beans

If you need maximum flexibility and don’t want to use the @KafkaClient support you can use the @KafkaClient annotation as qualifier for dependency injection of KafkaProducer instances.

Consider the following example:

Using a KafkaProducer directly

import io.micronaut.configuration.kafka.annotation.KafkaClient;
import io.micronaut.context.annotation.Requires;
import jakarta.inject.Singleton;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.concurrent.Future;

@Singleton
public class BookSender {

    private final Producer<String, Book> kafkaProducer;

    public BookSender(@KafkaClient("book-producer") Producer<String, Book> kafkaProducer) { // (1)
        this.kafkaProducer = kafkaProducer;
    }

    public Future<RecordMetadata> send(String author, Book book) {
        return kafkaProducer.send(new ProducerRecord<>("books", author, book)); // (2)
    }
}

Using a KafkaProducer directly

import io.micronaut.configuration.kafka.annotation.KafkaClient
import io.micronaut.context.annotation.Requires
import jakarta.inject.Singleton
import org.apache.kafka.clients.producer.Producer
import org.apache.kafka.clients.producer.ProducerRecord
import org.apache.kafka.clients.producer.RecordMetadata

import java.util.concurrent.Future

@Singleton
class BookSender {

    private final Producer<String, Book> kafkaProducer

    BookSender(@KafkaClient('book-producer') Producer<String, Book> kafkaProducer) { // (1)
        this.kafkaProducer = kafkaProducer
    }

    Future<RecordMetadata> send(String author, Book book) {
        kafkaProducer.send(new ProducerRecord<>('books', author, book)) // (2)
    }
}

Using a KafkaProducer directly

import io.micronaut.configuration.kafka.annotation.KafkaClient
import io.micronaut.context.annotation.Requires
import jakarta.inject.Singleton
import org.apache.kafka.clients.producer.Producer
import org.apache.kafka.clients.producer.ProducerRecord
import org.apache.kafka.clients.producer.RecordMetadata
import java.util.concurrent.Future

@Requires(property = "spec.name", value = "BookSenderTest")
@Singleton
class BookSender(
    @param:KafkaClient("book-producer") private val kafkaProducer: Producer<String, Book>) { // (1)

    fun send(author: String, book: Book): Future<RecordMetadata> {
        return kafkaProducer.send(ProducerRecord("books", author, book)) // (2)
    }
}

1	The `Producer` is dependency injected into the constructor. If not specified in configuration, the key and value serializer are inferred from the generic type arguments.
2	The `Producer` is used to send records

Note that there is no need to call the close() method to shut down the KafkaProducer, it is fully managed by Micronaut and will be shutdown when the application shuts down.

The previous example can be tested in JUnit with the following test:

Using a KafkaProducer directly

@Test
void testBookSender() {

    try (ApplicationContext ctx = ApplicationContext.run( // (1)
        Map.of("kafka.enabled", "true", "spec.name", "BookSenderTest")
    )) {
        BookSender bookSender = ctx.getBean(BookSender.class); // (2)
        Book book = new Book("The Stand");
        Future<RecordMetadata> stephenKing = bookSender.send("Stephen King", book);
        assertDoesNotThrow(() -> {
            RecordMetadata recordMetadata = stephenKing.get();
            assertEquals("books", recordMetadata.topic());
        });
    }
}

Using a KafkaProducer directly

void "test Book Sender"() {
    given:
    ApplicationContext ctx = ApplicationContext.run(  // (1)
        'kafka.enabled': true, 'spec.name': 'BookSenderTest'
    )
    BookSender bookSender = ctx.getBean(BookSender) // (2)
    Book book = new Book('The Stand')

    when:
    bookSender.send('Stephen King', book)
    Future<RecordMetadata> stephenKing = bookSender.send('Stephen King', book);
    def recordMetadata = stephenKing.get();

    then:
    noExceptionThrown()
    recordMetadata.topic() == 'books'

    cleanup:
    ctx.close()
}

Using a KafkaProducer directly

@Test
fun testBookSender() {
    ApplicationContext.run(mapOf( // (1)
        "kafka.enabled" to StringUtils.TRUE, "spec.name" to "BookSenderTest")).use { ctx ->
        val bookSender = ctx.getBean(BookSender::class.java) // (2)
        val book = Book("The Stand")
        bookSender.send("Stephen King", book)
        val stephenKing = bookSender.send("Stephen King", book)
        Assertions.assertDoesNotThrow {
            val recordMetadata = stephenKing.get()
            Assertions.assertEquals("books", recordMetadata.topic())
        }
    }

1	A Kafka docker container is used
2	The `BookSender` is retrieved from the ApplicationContext and a `ProducerRecord` sent

By using the KafkaProducer API directly you open up even more options if you require transactions (exactly-once delivery) or want control over when records are flushed etc.

5.5 Transactions

Transaction processing can be enabled by defining transactionalId on @KafkaClient, which will initialize the producer for transactional usage and wrap any send operation with a transaction demarcation.

Transactional Client

@KafkaClient(id = "my-client", transactionalId = "my-tx-id")
public interface TransactionalClient {
    // define client API
}

Transactional Client

@KafkaClient(id = "my-client", transactionalId = "my-tx-id")
interface TransactionalClient {
    // define client API
}

Transactional Client

@KafkaClient(id = "my-client", transactionalId = "my-tx-id")
interface TransactionalClient {
    // define client API
}

Alternative Kafka producer transactional code

producer.initTransactions();
try {
    producer.beginTransaction();
    producer.send(...);
    producer.commitTransaction();
} catch (Exception e) {
    producer.abortTransaction();
}

@KafkaClient beans are by default singleton. When using multiple threads, you must either synchronize access to the individual instance or declare the bean as @Prototype. Additionally, you can use random properties to your advantage so that each instance of your producer gets a different transactional ID.

Random transactional ID

@Prototype
@KafkaClient(id = "my-client", transactionalId = "my-tx-id-${random.uuid}")
public interface RandomTransactionalIdClient {
    // define client API
}

Random transactional ID

@Prototype
@KafkaClient(id = "my-client", transactionalId = 'my-tx-id-${random.uuid}')
interface RandomTransactionalIdClient {
    // define client API
}

Random transactional ID

@Prototype
@KafkaClient(id = "my-client", transactionalId = "my-tx-id-\${random.uuid}")
interface RandomTransactionalIdClient {
    // define client API
}

5.6 Running Kafka while testing and developing

Micronaut Test Resources simplifies running Kafka for local development and testing.

Micronaut Test Resources Kafka support will automatically start a Kafka container and provide the value of the kafka.bootstrap.servers property.

Micronaut Launch and CLI already apply Test Resources to your build when you select the kafka feature.

Micronaut Test Resources uses Test Containers under the hood. If you prefer to use Test Containers directly, you can create a Singleton Container and combine it with Micronaut Test TestPropertyProvider:

package io.micronaut.kafka.docs;

import io.micronaut.test.support.TestPropertyProvider;
import org.testcontainers.containers.KafkaContainer;
import org.testcontainers.utility.DockerImageName;

import java.util.Collections;
import java.util.Map;

/**
 * @see <a href="https://www.testcontainers.org/test_framework_integration/manual_lifecycle_control/#singleton-containers">Singleton containers</a>
 */
public abstract class AbstractKafkaTest implements TestPropertyProvider {

    static protected final KafkaContainer MY_KAFKA = new KafkaContainer(
        DockerImageName.parse("confluentinc/cp-kafka:latest")
    );

    @Override
    public Map<String, String> getProperties() {
        if (!MY_KAFKA.isRunning()) {
            MY_KAFKA.start();
        }
        return Collections.singletonMap(
            "kafka.bootstrap.servers", MY_KAFKA.getBootstrapServers()
        );
    }
}

package io.micronaut.kafka.docs

import io.micronaut.test.support.TestPropertyProvider
import org.testcontainers.containers.KafkaContainer
import org.testcontainers.utility.DockerImageName
import spock.lang.AutoCleanup
import spock.lang.Shared
import spock.lang.Specification

abstract class AbstractKafkaTest extends Specification implements TestPropertyProvider {

    @Shared
    @AutoCleanup
    KafkaContainer kafkaContainer = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:latest"))

    @Override
    Map<String, String> getProperties() {
        kafkaContainer.start()

        ["kafka.bootstrap.servers": kafkaContainer.getBootstrapServers()]
    }
}

package io.micronaut.kafka.docs

import io.micronaut.test.support.TestPropertyProvider
import org.testcontainers.containers.KafkaContainer
import org.testcontainers.utility.DockerImageName

/**
 * @see <a href="https://www.testcontainers.org/test_framework_integration/manual_lifecycle_control/#singleton-containers">Singleton containers</a>
 */
abstract class AbstractKafkaTest : TestPropertyProvider {

    companion object {
        var MY_KAFKA: KafkaContainer = KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:latest"))
    }

    override fun getProperties(): MutableMap<String, String> {
        if(!MY_KAFKA.isRunning) {
            MY_KAFKA.start()
        }
        return mutableMapOf(
            "kafka.bootstrap.servers" to MY_KAFKA.bootstrapServers
        )
    }
}

And then test:

package io.micronaut.kafka.docs;

import io.micronaut.configuration.kafka.annotation.KafkaClient;
import io.micronaut.configuration.kafka.annotation.KafkaListener;
import io.micronaut.configuration.kafka.annotation.OffsetReset;
import io.micronaut.configuration.kafka.annotation.Topic;
import io.micronaut.context.annotation.Property;
import io.micronaut.context.annotation.Requires;
import io.micronaut.test.extensions.junit5.annotation.MicronautTest;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInstance;

import static java.util.concurrent.TimeUnit.SECONDS;
import static org.awaitility.Awaitility.await;

@Property(name = "spec.name", value = "MyTest")
@MicronautTest
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
class MyTest extends AbstractKafkaTest {

    @Test
    void testKafkaRunning(MyProducer producer, MyConsumer consumer) {
        final String message = "hello";
        producer.produce(message);
        await().atMost(5, SECONDS).until(() -> message.equals(consumer.consumed));
    }

    @Requires(property = "spec.name", value = "MyTest")
    @KafkaClient
    interface MyProducer {
        @Topic("my-topic")
        void produce(String message);
    }

    @Requires(property = "spec.name", value = "MyTest")
    @KafkaListener(offsetReset = OffsetReset.EARLIEST)
    static class MyConsumer {
        String consumed;
        @Topic("my-topic")
        public void consume(String message) {
            consumed = message;
        }
    }
}

package io.micronaut.kafka.docs


import io.micronaut.configuration.kafka.annotation.KafkaClient
import io.micronaut.configuration.kafka.annotation.KafkaListener
import io.micronaut.configuration.kafka.annotation.OffsetReset
import io.micronaut.configuration.kafka.annotation.Topic
import io.micronaut.context.annotation.Property
import io.micronaut.context.annotation.Requires
import io.micronaut.test.extensions.spock.annotation.MicronautTest
import jakarta.inject.Inject
import spock.util.concurrent.PollingConditions

@MicronautTest
@Property(name = "spec.name", value = "MyTest")
class MyTest extends AbstractKafkaTest {

    @Inject
    MyProducer producer
    @Inject
    MyConsumer consumer

    PollingConditions conditions = new PollingConditions()

    void "test kafka running"() {
        given:
        String message = "hello"

        when:
        producer.produce(message)

        then:
        conditions.within(5) {
            consumer.consumed == message
        }
    }

    @Requires(property = "spec.name", value = "MyTest")
    @KafkaClient
    static interface MyProducer {
        @Topic("my-topic")
        void produce(String message)
    }

    @Requires(property = "spec.name", value = "MyTest")
    @KafkaListener(offsetReset = OffsetReset.EARLIEST)
    static class MyConsumer {
        String consumed

        @Topic("my-topic")
        void consume(String message) {
            consumed = message
        }
    }

}

package io.micronaut.kafka.docs

import io.micronaut.configuration.kafka.annotation.KafkaClient
import io.micronaut.configuration.kafka.annotation.KafkaListener
import io.micronaut.configuration.kafka.annotation.OffsetReset
import io.micronaut.configuration.kafka.annotation.Topic
import io.micronaut.context.annotation.Property
import io.micronaut.context.annotation.Requires
import io.micronaut.test.extensions.junit5.annotation.MicronautTest
import org.awaitility.Awaitility.await
import org.junit.jupiter.api.Test
import org.junit.jupiter.api.TestInstance
import java.util.concurrent.TimeUnit

@Property(name = "spec.name", value = "MyTest")
@MicronautTest
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
internal class MyTest : AbstractKafkaTest() {

    @Test
    fun testKafkaRunning(producer: MyProducer, consumer: MyConsumer) {
        val message = "hello"
        producer.produce(message)
        await().atMost(5, TimeUnit.SECONDS).until { consumer.consumed == message }
    }

    @Requires(property = "spec.name", value = "MyTest")
    @KafkaClient
    interface MyProducer {
        @Topic("my-topic")
        fun produce(message: String)
    }

    @Requires(property = "spec.name", value = "MyTest")
    @KafkaListener(offsetReset = OffsetReset.EARLIEST)
    class MyConsumer {
        var consumed: String? = null

        @Topic("my-topic")
        fun consume(message: String) {
            consumed = message
        }
    }
}

6 Kafka Consumers Using @KafkaListener

The quick start section presented a trivial example of what is possible with the @KafkaListener annotation.

Using the @KafkaListener annotation Micronaut will build a KafkaConsumer and start the poll loop by running the KafkaConsumer in a special consumer thread pool. You can configure the size of the thread pool based on the number of consumers in your application in application.yml as desired:

Configuring the consumer thread pool

micronaut.executors.consumer.type=fixed
micronaut.executors.consumer.nThreads=25

micronaut:
    executors:
        consumer:
            type: fixed
            nThreads: 25

[micronaut]
  [micronaut.executors]
    [micronaut.executors.consumer]
      type="fixed"
      nThreads=25

micronaut {
  executors {
    consumer {
      type = "fixed"
      nThreads = 25
    }
  }
}

{
  micronaut {
    executors {
      consumer {
        type = "fixed"
        nThreads = 25
      }
    }
  }
}

{
  "micronaut": {
    "executors": {
      "consumer": {
        "type": "fixed",
        "nThreads": 25
      }
    }
  }
}

KafkaConsumer instances are single threaded, hence for each @KafkaListener method you define a new thread is created to execute the poll loop.

You may wish to scale the number of consumers you have listening on a particular topic. There are several ways you may achieve this. You could for example run multiple instances of your application each containing a single consumer in each JVM.

Alternatively, you can also scale via threads. By setting the number of threads a particular consumer bean will create:

Scaling with Threads

@KafkaListener(groupId = "myGroup", threads = 10)

Scaling with Threads

@KafkaListener(groupId='myGroup', threads = 10)

Scaling with Threads

@KafkaListener(groupId = "myGroup", threads = 10)

The above example will create 10 KafkaConsumer instances, each running in a unique thread and participating in the myGroup consumer group.

@KafkaListener beans are by default singleton. When using multiple threads you must either synchronize access to local state or declare the bean as @Prototype.

You can also make your number of threads configurable by using threadsValue:

Dynamically Configuring Threads

@KafkaListener(groupId = "myGroup", threadsValue = "${my.thread.count}")

Dynamically Configuring Threads

@KafkaListener(groupId = 'myGroup', threadsValue = '${my.thread.count}')

Dynamically Configuring Threads

@KafkaListener(groupId = "myGroup", threadsValue = "\${my.thread.count}")

threads will be overridden by threadsValue if they are both set.

By default Micronaut will inspect the method signature of the method annotated with @Topic that will listen for ConsumerRecord instances and from the types infer an appropriate key and value Deserializer.

Applying Configuration

kafka.consumers.default.allow.auto.create.topics=true
kafka.consumers.product.bootstrap.servers=localhost:9098

kafka:
    consumers:
        default:
            allow.auto.create.topics: true
        product:
            bootstrap:
              servers: localhost:9098

[kafka]
  [kafka.consumers]
    [kafka.consumers.default]
      "allow.auto.create.topics"=true
    [kafka.consumers.product]
      [kafka.consumers.product.bootstrap]
        servers="localhost:9098"

kafka {
  consumers {
    'default' {
      allow.auto.create.topics = true
    }
    product {
      bootstrap {
        servers = "localhost:9098"
      }
    }
  }
}

{
  kafka {
    consumers {
      default {
        "allow.auto.create.topics" = true
      }
      product {
        bootstrap {
          servers = "localhost:9098"
        }
      }
    }
  }
}

{
  "kafka": {
    "consumers": {
      "default": {
        "allow.auto.create.topics": true
      },
      "product": {
        "bootstrap": {
          "servers": "localhost:9098"
        }
      }
    }
  }
}

Any property in the ConsumerConfig class can be set for all @KafkaListener beans based on the . The above example will enable the consumer to create a topic if it doesn’t exist for the default (@KafkaListener) client and set a custom bootstrap server for the product client (@KafkaListener(value = "product"))

See the guide for Testing Kafka Listener using Testcontainers with the Micronaut Framework to learn more.

6.1 Defining @KafkaListener Methods

The @KafkaListener annotation examples up until now have been relatively trivial, but Micronaut offers a lot of flexibility when it comes to the types of method signatures you can define.

The following sections detail examples of supported use cases.

Specifying Topics

The @Topic annotation can be used at the method or the class level to specify which topics to be listened for.

Care needs to be taken when using @Topic at the class level because every public method of the class annotated with @KafkaListener will become a Kafka consumer, which may be undesirable.

You can specify multiple topics to listen for:

Specifying Multiple Topics

@Topic({"fun-products", "awesome-products"})
public void receiveMultiTopics(@KafkaKey String brand, String name) {
    LOG.info("Got Product - {} by {}", name, brand);
}

Specifying Multiple Topics

@Topic(['fun-products', 'awesome-products'])
void receiveMultiTopics(@KafkaKey String brand, String name) {
    log.info("Got Product - {} by {}", name, brand)
}

Specifying Multiple Topics

@Topic("fun-products", "awesome-products")
fun receiveMultiTopics(@KafkaKey brand: String, name: String) {
    LOG.info("Got Product - {} by {}", name, brand)
}

You can also specify one or many regular expressions to listen for:

Using regular expressions to match Topics

@Topic(patterns="products-\\w+")
public void receivePatternTopics(@KafkaKey String brand, String name) {
    LOG.info("Got Product - {} by {}", name, brand);
}

Using regular expressions to match Topics

@Topic(patterns='products-\\w+')
void receivePatternTopics(@KafkaKey String brand, String name) {
    log.info("Got Product - {} by {}", name, brand)
}

Using regular expressions to match Topics

@Topic(patterns = ["products-\\w+"])
fun receivePatternTopics(@KafkaKey brand: String, name: String) {
    LOG.info("Got Product - {} by {}", name, brand)
}

Available Annotations

There are a number of annotations available that allow you to specify how a method argument is bound.

The following table summarizes the annotations and their purpose, with an example:

Table 1. Kafka Messaging Annotations
Annotation	Description	Example
@MessageBody	Allows explicitly indicating the body of the message	`@MessageBody Product product`
@MessageHeader	Allows binding a parameter to a message header	`@MessageHeader("X-My-Header") String myHeader`
@KafkaKey	Allows binding a parameter to the message key	`@KafkaKey String messageKey`
@KafkaPartition	Allows binding a parameter to the partition the message was received from	`@KafkaPartition Integer partition`

For example, you can use the @MessageHeader annotation to bind a parameter value from a header contained within a ConsumerRecord.

Topics, Partitions and Offsets

If you want a reference to the topic, partition or offset it is a simple matter of defining a parameter for each.

The following table summarizes example parameters and how they related to the ConsumerRecord being processed:

Table 2. @KafkaListener Method Parameters
Parameter	Description
`String topic`	The name of the topic
`long offset`	The offset of the `ConsumerRecord`
`int partition`	The partition of the `ConsumerRecord`
`long timestamp`	The timestamp of the `ConsumerRecord`

As an example, following listener method will receive all of the above mentioned parameters: