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

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

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

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

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

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

@KafkaClient(batch = true)
public interface BookClient {

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

@KafkaClient(batch = true)
interface BookClient {

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

@KafkaClient(batch = true)
interface BookClient {

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

@KafkaClient(batch = true)
public interface BookClient {

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

@KafkaClient(batch = true)
interface BookClient {

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

@KafkaClient(batch = true)
interface BookClient {

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

@KafkaClient(batch = true)
public interface BookClient {

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

@KafkaClient(batch = true)
interface BookClient {

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

@KafkaClient(batch = true)
interface BookClient {

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

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)
    }
}

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)
    }
}

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)
    }
}

@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());
        });
    }
}

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()
}

@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())
        }
    }

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

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

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

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
        )
    }
}

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
        }
    }
}

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

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

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

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

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

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

package io.micronaut.kafka.docs.seek.aware;

import io.micronaut.configuration.kafka.ConsumerSeekAware;
import io.micronaut.configuration.kafka.annotation.*;
import io.micronaut.configuration.kafka.seek.*;
import io.micronaut.context.annotation.Requires;
import io.micronaut.kafka.docs.Product;
import org.apache.kafka.common.TopicPartition;
import java.util.*;

@KafkaListener
@Requires(property = "spec.name", value = "ConsumerSeekAwareTest")
public class ProductListener implements ConsumerSeekAware { // (1)

    List<Product> processed = new ArrayList<>();

    public ProductListener(ProductListenerConfiguration config) {
        // ...
    }

    @Topic("wonderful-products")
    void receive(Product product) {
        processed.add(product);
    }

    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) { // (2)
        // save offsets here
    }

    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions, KafkaSeeker seeker) { // (3)
        // seek to offset here
        partitions.stream().map(tp -> KafkaSeekOperation.seek(tp, 1)).forEach(seeker::perform);
    }
}

package io.micronaut.kafka.docs.seek.aware

import io.micronaut.configuration.kafka.ConsumerSeekAware
import io.micronaut.configuration.kafka.annotation.*
import io.micronaut.configuration.kafka.seek.*
import io.micronaut.context.annotation.Requires
import io.micronaut.kafka.docs.Product
import org.apache.kafka.common.TopicPartition

@KafkaListener
@Requires(property = "spec.name", value = "ConsumerSeekAwareSpec")
class ProductListener implements ConsumerSeekAware { // (1)

    List<Product> processed = []

    ProductListener(ProductListenerConfiguration config) {
        // ...
    }

    @Topic("wonderful-products")
    void receive(Product product) {
        processed << product
    }

    @Override
    void onPartitionsRevoked(Collection<TopicPartition> partitions) { // (2)
        // save offsets here
    }

    @Override
    void onPartitionsAssigned(Collection<TopicPartition> partitions, KafkaSeeker seeker) { // (3)
        // seek to offset here
        partitions.collect { KafkaSeekOperation.seek(it, 1) }.each(seeker.&perform)
    }
}

package io.micronaut.kafka.docs.seek.aware

import io.micronaut.configuration.kafka.ConsumerSeekAware
import io.micronaut.configuration.kafka.annotation.*
import io.micronaut.configuration.kafka.seek.*
import io.micronaut.context.annotation.Requires
import io.micronaut.kafka.docs.Product
import org.apache.kafka.common.TopicPartition

@KafkaListener
@Requires(property = "spec.name", value = "ConsumerSeekAwareTest")
class ProductListener constructor(config: ProductListenerConfiguration) : ConsumerSeekAware { // (1)

    var processed: MutableList<Product> = mutableListOf()

    @Topic("wonderful-products")
    fun receive(product: Product) {
        processed.add(product)
    }

    override fun onPartitionsRevoked(partitions: Collection<TopicPartition>) { // (2)
        // save offsets here
    }

    override fun onPartitionsAssigned(partitions: Collection<TopicPartition>, seeker: KafkaSeeker) { // (3)
        // seek to offset here
        partitions.stream().map { tp -> KafkaSeekOperation.seek(tp, 1) }.forEach(seeker::perform)
    }
}

package io.micronaut.kafka.docs.seek.rebalance;

import io.micronaut.configuration.kafka.ConsumerAware;
import io.micronaut.configuration.kafka.annotation.*;
import io.micronaut.context.annotation.Requires;
import io.micronaut.kafka.docs.Product;
import io.micronaut.core.annotation.NonNull;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import java.util.*;

@KafkaListener(offsetReset = OffsetReset.EARLIEST)
@Requires(property = "spec.name", value = "ConsumerRebalanceListenerTest")
public class ProductListener implements ConsumerRebalanceListener, ConsumerAware {

    List<Product> processed = new ArrayList<>();
    private Consumer consumer;

    public ProductListener(ProductListenerConfiguration config) {
        // ...
    }

    @Override
    public void setKafkaConsumer(@NonNull Consumer consumer) { // (1)
        this.consumer = consumer;
    }

    @Topic("fantastic-products")
    void receive(Product product) {
        processed.add(product);
    }

    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) { // (2)
        // save offsets here
    }

    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) { // (3)
        // seek to offset here
        for (TopicPartition partition : partitions) {
            consumer.seek(partition, 1);
        }
    }
}

package io.micronaut.kafka.docs.seek.rebalance

import io.micronaut.configuration.kafka.ConsumerAware
import io.micronaut.configuration.kafka.annotation.*
import io.micronaut.context.annotation.Requires
import io.micronaut.core.annotation.NonNull
import io.micronaut.kafka.docs.Product
import org.apache.kafka.clients.consumer.*
import org.apache.kafka.common.TopicPartition

@KafkaListener(offsetReset = OffsetReset.EARLIEST)
@Requires(property = "spec.name", value = "ConsumerRebalanceListenerSpec")
class ProductListener implements ConsumerRebalanceListener, ConsumerAware {

    List<Product> processed = []
    private Consumer consumer

    ProductListener(ProductListenerConfiguration config) {
        // ...
    }

    @Override
    void setKafkaConsumer(@NonNull Consumer consumer) { // (1)
        this.consumer = consumer
    }

    @Topic("fantastic-products")
    void receive(Product product) {
        processed << product
    }

    @Override
    void onPartitionsRevoked(Collection<TopicPartition> partitions) { // (2)
        // save offsets here
    }

    @Override
    void onPartitionsAssigned(Collection<TopicPartition> partitions) { // (3)
        // seek to offset here
        for (TopicPartition partition : partitions) {
            consumer.seek(partition, 1)
        }
    }
}

package io.micronaut.kafka.docs.seek.rebalance

import io.micronaut.configuration.kafka.ConsumerAware
import io.micronaut.configuration.kafka.annotation.*
import io.micronaut.context.annotation.Requires
import io.micronaut.kafka.docs.Product
import org.apache.kafka.clients.consumer.*
import org.apache.kafka.common.TopicPartition

@KafkaListener(offsetReset = OffsetReset.EARLIEST)
@Requires(property = "spec.name", value = "ConsumerRebalanceListenerTest")
class ProductListener constructor(config: ProductListenerConfiguration) : ConsumerRebalanceListener, ConsumerAware<Any?, Any?> {

    var processed: MutableList<Product> = mutableListOf()
    private var consumer: Consumer<*, *>? = null

    override fun setKafkaConsumer(consumer: Consumer<Any?, Any?>?) { // (1)
        this.consumer = consumer
    }

    @Topic("fantastic-products")
    fun receive(product: Product) {
        processed.add(product)
    }

    override fun onPartitionsRevoked(partitions: Collection<TopicPartition>) { // (2)
        // save offsets here
    }

    override fun onPartitionsAssigned(partitions: Collection<TopicPartition>) { // (3)
        // seek to offset here
        for (partition in partitions) {
            consumer!!.seek(partition, 1)
        }
    }
}

package io.micronaut.kafka.docs.seek.ops;

import io.micronaut.configuration.kafka.annotation.*;
import io.micronaut.configuration.kafka.seek.*;
import io.micronaut.context.annotation.*;
import io.micronaut.kafka.docs.Product;
import org.apache.kafka.common.TopicPartition;
import java.util.*;

@KafkaListener(offsetReset = OffsetReset.EARLIEST, properties = @Property(name = "max.poll.records", value = "1"))
@Requires(property = "spec.name", value = "KafkaSeekOperationsTest")
public class ProductListener {

    List<Product> processed = new ArrayList<>();

    public ProductListener(ProductListenerConfiguration config) {
        // ...
    }

    @Topic("amazing-products")
    void receive(Product product, KafkaSeekOperations ops) { // (1)
        processed.add(product);
        ops.defer(KafkaSeekOperation.seekToEnd(new TopicPartition("amazing-products", 0))); // (2)
    }
}

package io.micronaut.kafka.docs.seek.ops

import io.micronaut.configuration.kafka.annotation.*
import io.micronaut.configuration.kafka.seek.*
import io.micronaut.context.annotation.*
import io.micronaut.kafka.docs.Product
import org.apache.kafka.common.TopicPartition

@KafkaListener(offsetReset = OffsetReset.EARLIEST, properties = @Property(name = "max.poll.records", value = "1"))
@Requires(property = "spec.name", value = "KafkaSeekOperationsSpec")
class ProductListener {

    List<Product> processed = []

    ProductListener(ProductListenerConfiguration config) {
        // ...
    }

    @Topic("amazing-products")
    void receive(Product product, KafkaSeekOperations ops) { // (1)
        processed << product
        ops.defer(KafkaSeekOperation.seekToEnd(new TopicPartition("amazing-products", 0))); // (2)
    }
}

package io.micronaut.kafka.docs.seek.ops

import io.micronaut.configuration.kafka.annotation.*
import io.micronaut.configuration.kafka.seek.*
import io.micronaut.context.annotation.*
import io.micronaut.kafka.docs.Product
import org.apache.kafka.common.TopicPartition

@KafkaListener(offsetReset = OffsetReset.EARLIEST, properties = [Property(name = "max.poll.records", value = "1")])
@Requires(property = "spec.name", value = "KafkaSeekOperationsTest")
class ProductListener constructor(config: ProductListenerConfiguration) {

    var processed: MutableList<Product> = mutableListOf()

    @Topic("amazing-products")
    fun receive(product: Product, ops: KafkaSeekOperations) { // (1)
        processed.add(product)
        ops.defer(KafkaSeekOperation.seekToBeginning(TopicPartition("amazing-products", 0))) // (2)
    }
}

import io.micronaut.configuration.kafka.annotation.KafkaListener;
import io.micronaut.configuration.kafka.annotation.Topic;
import io.micronaut.context.annotation.Requires;
import org.slf4j.Logger;
import reactor.core.publisher.Flux;

import java.util.List;

import static org.slf4j.LoggerFactory.getLogger;

@KafkaListener(batch = true) // (1)
public class BookListener {

@Topic("all-the-books")
public void receiveList(List<Book> books) { // (2)
    for (Book book : books) {
        LOG.info("Got Book = {}", book.title()); // (3)
    }
}

}

import io.micronaut.configuration.kafka.annotation.KafkaListener
import io.micronaut.configuration.kafka.annotation.Topic
import io.micronaut.context.annotation.Requires
import reactor.core.publisher.Flux

@KafkaListener(batch = true) // (1)
@Slf4j
class BookListener {

@Topic("all-the-books")
void receiveList(List<Book> books) { // (2)
    for (Book book : books) {
        log.info("Got Book = {}", book.title) // (3)
    }
}

}

import io.micronaut.configuration.kafka.annotation.KafkaListener
import io.micronaut.configuration.kafka.annotation.Topic
import io.micronaut.context.annotation.Requires
import org.slf4j.LoggerFactory.getLogger
import reactor.core.publisher.Flux
import java.util.*

@KafkaListener(batch = true) // (1)
class BookListener {

@Topic("all-the-books")
fun receiveList(books: List<Book>) { // (2)
    for (book in books) {
        LOG.info("Got Book = {}", book.title) // (3)
    }
}

}

@Topic("sendto-products") // (1)
@SendTo("product-quantities") // (2)
public int receive(@KafkaKey String brand, Product product) {
    LOG.info("Got Product - {} by {}", product.name(), brand);
    return product.quantity(); // (3)
}

@Topic("sendto-products") // (1)
@SendTo("product-quantities") // (2)
int receive(@KafkaKey String brand, Product product) {
    log.info("Got Product - {} by {}", product.name, brand)
    product.quantity // (3)
}

@Topic("sendto-products") // (1)
@SendTo("product-quantities") // (2)
fun receive(@KafkaKey brand: String?, product: Product): Int {
    LOG.info("Got Product - {} by {}", product.name, brand)
    return product.quantity // (3)
}

@Topic("sendto-products") // (1)
@SendTo("product-quantities") // (2)
public Mono<Integer> receiveProduct(@KafkaKey String brand,
                                    Mono<Product> productSingle) {

    return productSingle.map(product -> {
        LOG.info("Got Product - {} by {}", product.name(), brand);
        return product.quantity(); // (3)
    });
}

@Topic("sendto-products") // (1)
@SendTo("product-quantities") // (2)
Mono<Integer> receiveProduct(@KafkaKey String brand, Mono<Product> productSingle) {
    productSingle.map(product -> {
        log.info("Got Product - {} by {}", product.name, brand)
        product.quantity // (3)
    })
}

@Topic("sendto-products") // (1)
@SendTo("product-quantities") // (2)
fun receiveProduct(@KafkaKey brand: String?, productSingle: Mono<Product>): Mono<Int> {
    return productSingle.map(Function<Product, Int> { product: Product ->
        LOG.info("Got Product - {} by {}", product.name, brand)
        product.quantity // (3)
    })
}

@KafkaListener(
    offsetReset = OffsetReset.EARLIEST,
    producerClientId = "word-counter-producer", // (1)
    producerTransactionalId = "tx-word-counter-id", // (2)
    offsetStrategy = OffsetStrategy.SEND_TO_TRANSACTION, // (3)
    isolation = IsolationLevel.READ_COMMITTED // (4)
)
public class WordCounter {

    @Topic("tx-incoming-strings")
    @SendTo("my-words-count")
    List<KafkaMessage<byte[], Integer>> wordsCounter(String string) {
        return Stream.of(string.split("\\s+"))
            .collect(Collectors.groupingBy(Function.identity(), Collectors.summingInt(i -> 1)))
            .entrySet()
            .stream()
            .map(e -> KafkaMessage.Builder.<byte[], Integer>withBody(e.getValue()).key(e.getKey().getBytes()).build())
            .toList();
    }
}

@KafkaListener(
        offsetReset = OffsetReset.EARLIEST,
        producerClientId = 'word-counter-producer', // (1)
        producerTransactionalId = 'tx-word-counter-id', // (2)
        offsetStrategy = OffsetStrategy.SEND_TO_TRANSACTION, // (3)
        isolation = IsolationLevel.READ_COMMITTED // (4)
)
class WordCounter {

    @Topic('tx-incoming-strings')
    @SendTo('my-words-count')
    List<KafkaMessage<byte[], Integer>> wordsCounter(String string) {
        string.split("\\s+")
            .groupBy()
            .collect { key, instanceList ->
                KafkaMessage.Builder.withBody(instanceList.size()).key(key.bytes).build()
            }
    }
}

@KafkaListener(
    offsetReset = OffsetReset.EARLIEST,
    producerClientId = "word-counter-producer", // (1)
    producerTransactionalId = "tx-word-counter-id", // (2)
    offsetStrategy = OffsetStrategy.SEND_TO_TRANSACTION, // (3)
    isolation = IsolationLevel.READ_COMMITTED // (4)
)
class WordCounter {

    @Topic("tx-incoming-strings")
    @SendTo("my-words-count")
    fun wordsCounter(string: String) = string
        .split(Regex("\\s+"))
        .groupBy { it }
        .map { KafkaMessage.Builder.withBody<ByteArray, Int>(it.value.size).key(it.key.toByteArray()).build() }
}

import io.micronaut.context.annotation.Bean;
import io.micronaut.context.annotation.Factory;
import io.micronaut.context.annotation.Requires;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.CreateTopicsOptions;
import org.apache.kafka.clients.admin.NewTopic;

@Requires(bean = AdminClient.class)
@Factory
public class MyTopicFactory {

    @Bean
    CreateTopicsOptions options() {
        return new CreateTopicsOptions().timeoutMs(5000).validateOnly(true).retryOnQuotaViolation(false);
    }

    @Bean
    NewTopic topic1() {
        return new NewTopic("my-new-topic-1", 1, (short) 1);
    }

    @Bean
    NewTopic topic2() {
        return new NewTopic("my-new-topic-2", 2, (short) 1);
    }
}

import io.micronaut.context.annotation.Bean
import io.micronaut.context.annotation.Factory
import io.micronaut.context.annotation.Requires
import org.apache.kafka.clients.admin.AdminClient
import org.apache.kafka.clients.admin.CreateTopicsOptions
import org.apache.kafka.clients.admin.NewTopic

@Requires(bean = AdminClient)
@Factory
class MyTopicFactory {

    @Bean
    CreateTopicsOptions options() {
        new CreateTopicsOptions().timeoutMs(5000).validateOnly(true).retryOnQuotaViolation(false)
    }

    @Bean
    NewTopic topic1() {
        new NewTopic("my-new-topic-1", 1, (short) 1)
    }

    @Bean
    NewTopic topic2() {
        new NewTopic("my-new-topic-2", 2, (short) 1)
    }
}

import io.micronaut.context.annotation.Bean
import io.micronaut.context.annotation.Factory
import io.micronaut.context.annotation.Requires
import org.apache.kafka.clients.admin.AdminClient
import org.apache.kafka.clients.admin.CreateTopicsOptions
import org.apache.kafka.clients.admin.NewTopic

@Requires(bean = AdminClient::class)
@Factory
class MyTopicFactory {

    @Bean
    fun options(): CreateTopicsOptions {
        return CreateTopicsOptions().timeoutMs(5000).validateOnly(true).retryOnQuotaViolation(false)
    }

    @Bean
    fun topic1(): NewTopic {
        return NewTopic("my-new-topic-1", 1, 1)
    }

    @Bean
    fun topic2(): NewTopic {
        return NewTopic("my-new-topic-2", 2, 1)
    }
}

    boolean areNewTopicsDone(KafkaNewTopics newTopics) {
        return newTopics.getResult().all().isDone();
    }

    boolean areNewTopicsDone(KafkaNewTopics newTopics) {
        newTopics.result.all().done
    }

    fun areNewTopicsDone(newTopics: KafkaNewTopics): Boolean {
        return newTopics.result.all().isDone
    }

import io.micronaut.context.annotation.Replaces;
import io.micronaut.context.annotation.Requires;
import io.micronaut.core.util.StringUtils;
import jakarta.inject.Singleton;

@Requires(property = "kafka.enabled", notEquals = StringUtils.TRUE, defaultValue = StringUtils.TRUE) // (1)
@Replaces(MessageClient.class) // (2)
@Singleton
public class MessageClientFallback implements MessageClient { // (3)

    @Override
    public void send(String message) {
        throw new UnsupportedOperationException(); // (4)
    }
}

import io.micronaut.context.annotation.Replaces
import io.micronaut.context.annotation.Requires
import io.micronaut.core.util.StringUtils
import jakarta.inject.Singleton

@Requires(property = 'kafka.enabled', notEquals = StringUtils.TRUE, defaultValue = StringUtils.TRUE) // (1)
@Replaces(MessageClient.class) // (2)
@Singleton
class MessageClientFallback implements MessageClient { // (3)

    @Override
    void send(String message) {
        throw new UnsupportedOperationException() // (4)
    }
}

import io.micronaut.context.annotation.Replaces
import io.micronaut.context.annotation.Requires
import io.micronaut.core.util.StringUtils
import jakarta.inject.Singleton

@Requires(property = "kafka.enabled", notEquals = StringUtils.TRUE, defaultValue = StringUtils.TRUE) // (1)
@Replaces(MessageClient::class) // (2)
@Singleton
class MessageClientFallback : MessageClient { // (3)

    override fun send(message: String) {
        throw UnsupportedOperationException() // (4)
    }
}

import io.micronaut.configuration.kafka.streams.InteractiveQueryService;
import io.micronaut.context.annotation.Requires;
import jakarta.inject.Singleton;
import org.apache.kafka.streams.state.QueryableStoreTypes;
import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;

import java.util.Optional;

/**
 * Example service that uses the InteractiveQueryService in a reusable way.  This is only intended as an example.
 */
@Singleton
public class InteractiveQueryServiceExample {

    private final InteractiveQueryService interactiveQueryService;

    public InteractiveQueryServiceExample(InteractiveQueryService interactiveQueryService) {
        this.interactiveQueryService = interactiveQueryService;
    }

    /**
     * Method to get the word state store and word count from the store using the interactive query service.
     *
     * @param stateStore the name of the state store ie "foo-store"
     * @param word       the key to get, in this case the word as the stream and ktable have been grouped by word
     * @return the Long count of the word in the store
     */
    public Long getWordCount(String stateStore, String word) {
        Optional<ReadOnlyKeyValueStore<String, Long>> queryableStore = interactiveQueryService.getQueryableStore(
            stateStore, QueryableStoreTypes.keyValueStore());
        return queryableStore.map(kvReadOnlyKeyValueStore ->
            kvReadOnlyKeyValueStore.get(word)).orElse(0L);
    }

    /**
     * Method to get byte array from a state store using the interactive query service.
     *
     * @param stateStore the name of the state store ie "bar-store"
     * @param blobName   the key to get, in this case the name of the blob
     * @return the byte[] stored in the state store
     */
    public byte[] getBytes(String stateStore, String blobName) {
        Optional<ReadOnlyKeyValueStore<String, byte[]>> queryableStore = interactiveQueryService.getQueryableStore(
            stateStore, QueryableStoreTypes.keyValueStore());
        return queryableStore.map(stringReadOnlyKeyValueStore ->
            stringReadOnlyKeyValueStore.get(blobName)).orElse(null);
    }

    /**
     * Method to get value V by key K.
     *
     * @param stateStore the name of the state store ie "baz-store"
     * @param name       the key to get
     * @return the value of type V stored in the state store
     */
    public <K, V> V getGenericKeyValue(String stateStore, K name) {
        Optional<ReadOnlyKeyValueStore<K, V>> queryableStore = interactiveQueryService.getQueryableStore(
            stateStore, QueryableStoreTypes.<K, V>keyValueStore());
        return queryableStore.map(kvReadOnlyKeyValueStore ->
            kvReadOnlyKeyValueStore.get(name)).orElse(null);
    }
}

import io.micronaut.configuration.kafka.streams.InteractiveQueryService
import io.micronaut.context.annotation.Requires;
import jakarta.inject.Singleton;
import org.apache.kafka.streams.state.QueryableStoreTypes;
import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;

/**
 * Example service that uses the InteractiveQueryService in a reusable way.  This is only intended as an example.
 */
@Singleton
class InteractiveQueryServiceExample {

    private final InteractiveQueryService interactiveQueryService;

    InteractiveQueryServiceExample(InteractiveQueryService interactiveQueryService) {
        this.interactiveQueryService = interactiveQueryService;
    }

    /**
     * Method to get the word state store and word count from the store using the interactive query service.
     *
     * @param stateStore the name of the state store ie "foo-store"
     * @param word       the key to get, in this case the word as the stream and ktable have been grouped by word
     * @return the Long count of the word in the store
     */
    Long getWordCount(String stateStore, String word) {
        Optional<ReadOnlyKeyValueStore<String, Long>> queryableStore = interactiveQueryService.getQueryableStore(
                stateStore, QueryableStoreTypes.keyValueStore());
        return queryableStore.map(kvReadOnlyKeyValueStore ->
                kvReadOnlyKeyValueStore.get(word)).orElse(0L);
    }

    /**
     * Method to get byte array from a state store using the interactive query service.
     *
     * @param stateStore the name of the state store ie "bar-store"
     * @param blobName   the key to get, in this case the name of the blob
     * @return the byte[] stored in the state store
     */
    byte[] getBytes(String stateStore, String blobName) {
        Optional<ReadOnlyKeyValueStore<String, byte[]>> queryableStore = interactiveQueryService.getQueryableStore(
                stateStore, QueryableStoreTypes.keyValueStore());
        return queryableStore.map(stringReadOnlyKeyValueStore ->
                stringReadOnlyKeyValueStore.get(blobName)).orElse(null);
    }

    /**
     * Method to get value V by key K.
     *
     * @param stateStore the name of the state store ie "baz-store"
     * @param name       the key to get
     * @return the value of type V stored in the state store
     */
    <K, V> V getGenericKeyValue(String stateStore, K name) {
        Optional<ReadOnlyKeyValueStore<K, V>> queryableStore = interactiveQueryService.getQueryableStore(
                stateStore, QueryableStoreTypes.<K, V>keyValueStore());
        return queryableStore.map(kvReadOnlyKeyValueStore ->
                kvReadOnlyKeyValueStore.get(name)).orElse(null);
    }
}

import io.micronaut.configuration.kafka.streams.InteractiveQueryService
import io.micronaut.context.annotation.Requires
import jakarta.inject.Singleton
import org.apache.kafka.streams.state.QueryableStoreTypes
import org.apache.kafka.streams.state.ReadOnlyKeyValueStore

/**
 * Example service that uses the InteractiveQueryService in a reusable way.  This is only intended as an example.
 */
@Singleton
class InteractiveQueryServiceExample(private val interactiveQueryService: InteractiveQueryService) {
    /**
     * Method to get the word state store and word count from the store using the interactive query service.
     *
     * @param stateStore the name of the state store ie "foo-store"
     * @param word       the key to get, in this case the word as the stream and ktable have been grouped by word
     * @return the Long count of the word in the store
     */
    fun getWordCount(stateStore: String, word: String): Long {
        val queryableStore = interactiveQueryService.getQueryableStore(
            stateStore, QueryableStoreTypes.keyValueStore<String, Long>())
        return queryableStore.map { kvReadOnlyKeyValueStore: ReadOnlyKeyValueStore<String, Long> ->
            kvReadOnlyKeyValueStore[word] }.orElse(0L)
    }

    /**
     * Method to get byte array from a state store using the interactive query service.
     *
     * @param stateStore the name of the state store ie "bar-store"
     * @param blobName   the key to get, in this case the name of the blob
     * @return the byte[] stored in the state store
     */
    fun getBytes(stateStore: String, blobName: String): ByteArray? {
        val queryableStore = interactiveQueryService.getQueryableStore(
            stateStore, QueryableStoreTypes.keyValueStore<String, ByteArray>())
        return queryableStore.map { stringReadOnlyKeyValueStore: ReadOnlyKeyValueStore<String, ByteArray> ->
            stringReadOnlyKeyValueStore[blobName] }.orElse(null)
    }

    /**
     * Method to get value V by key K.
     *
     * @param stateStore the name of the state store ie "baz-store"
     * @param name       the key to get
     * @return the value of type V stored in the state store
     */
    fun <K, V> getGenericKeyValue(stateStore: String, name: K): V {
        val queryableStore = interactiveQueryService.getQueryableStore(
            stateStore, QueryableStoreTypes.keyValueStore<K, V>())
        return queryableStore.map { kvReadOnlyKeyValueStore: ReadOnlyKeyValueStore<K, V> ->
            kvReadOnlyKeyValueStore[name] }.orElse(null)
    }
}