@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)
interface BookClient {

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

@KafkaClient(batch = true)
interface BookClient {

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

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

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")
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 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.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', 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

@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.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

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