Kafka

The Kafka module provides declarative integration with Apache Kafka: reading messages through @KafkaListener, sending messages through @KafkaPublisher, serialization, deserialization, transactions, processing errors, and telemetry.

Apache Kafka is a distributed event streaming platform. Applications write events to a topic, while other applications read them through a consumer group or directly assigned partitions. Kora creates the required Consumer and Producer at compile time, binds them to the dependency graph, and lets most of the contract be described through method signatures.

For a step-by-step walkthrough before the reference details, see Kafka Messaging.

Dependency

Java Kotlin

Dependency build.gradle:

annotationProcessor "io.koraframework:annotation-processors" //(1)!
implementation "io.koraframework:kafka"

1. The annotation processor generates consumer containers and publisher implementations at compile time. Without it neither @KafkaListener nor @KafkaPublisher produces anything and the graph fails with a missing dependency.

Module:

@KoraApp
public interface Application extends KafkaModule { }

Dependency build.gradle.kts:

ksp("io.koraframework:symbol-processors:2.0.0.RC1") //(1)!
implementation("io.koraframework:kafka")

1. The KSP processor generates consumer containers and publisher implementations at compile time. Without it neither @KafkaListener nor @KafkaPublisher produces anything and the graph fails with a missing dependency.

Module:

@KoraApp
interface Application : KafkaModule

The module is built on the official Apache Kafka client and uses its Consumer, Producer, ConsumerRecord, ProducerRecord, Serializer, and Deserializer contracts directly, so any driver setting is available through driverProperties.

Consumer

Consumer reads records from a topic and passes them to an application method. Kora creates the consumer container, calls poll(), applies deserialization, invokes the handler, and commits the offset unless the method signature requires manual Consumer control.

Creating a Consumer requires using the @KafkaListener annotation over a method:

Java Kotlin

@Component
final class ConsumerService {

    @KafkaListener("kafka.someConsumer")
    void process(String key, String value) {
        // my code
    }
}

@Component
class ConsumerService {

    @KafkaListener("kafka.someConsumer")
    fun process(key: String, value: String) {
        // my code
    }
}

The @KafkaListener annotation parameter points to the Consumer configuration path. The class that declares the method must itself be a graph component, because the generated container receives it as a dependency.

In case you need different behavior for different topics, it is possible to create several such containers, each with its own individual configuration. It looks like this:

Java Kotlin

@Component
final class ConsumerService {

    @KafkaListener("kafka.someConsumer1")
    void processFirst(String key, String value) {
        // some handler code
    }

    @KafkaListener("kafka.someConsumer2")
    void processSecond(String key, String value) {
        // some handler code
    }
}

@Component
class ConsumerService {

    @KafkaListener("kafka.someConsumer1")
    fun processFirst(key: String, value: String) {
        // some handler code
    }

    @KafkaListener("kafka.someConsumer2")
    fun processSecond(key: String, value: String) {
        // some handler code
    }
}

The value in the annotation indicates which part of the configuration file should be used. Conceptually, it is similar to @ConfigSource: the annotation value selects the configuration branch for a specific container.

Configuration

Configuration describes the settings of a particular @KafkaListener and an example for the configuration at path kafka.someConsumer is given below.

Basic configuration parameters:

Hocon YAML

kafka {
    someConsumer {
        topics = ["topic1", "topic2"] //(1)!
        offset = "latest" //(2)!
        pollTimeout = "5s" //(3)!
        threads = 1 //(4)!
        driverProperties { //(5)!
            "bootstrap.servers": "localhost:9093"
            "group.id": "my-group-id"
        }
    }
}

1. List of topics to subscribe to (required to specify either topics or topicsPattern)
2. Initial read position (default: latest). Allowed values: earliest, latest, or time offset (e.g. 5m)
3. Maximum time to wait for messages (default: 5s)
4. Number of threads for the consumer (default: 1)
5. Official Kafka Consumer Properties (required, no default)

kafka:
  someConsumer:
    topics:
      - "topic1"
      - "topic2" #(1)!
    offset: "latest" #(2)!
    pollTimeout: "5s" #(3)!
    threads: 1 #(4)!
    driverProperties: #(5)!
      "bootstrap.servers": "localhost:9093"
      "group.id": "my-group-id"

1. List of topics to subscribe to (required to specify either topics or topicsPattern)
2. Initial read position (default: latest). Allowed values: earliest, latest, or time offset (e.g. 5m)
3. Maximum time to wait for messages (default: 5s)
4. Number of threads for the consumer (default: 1)
5. Official Kafka Consumer Properties (required, no default)

Full Configuration

Example of the complete configuration described in the KafkaListenerConfig class (default or example values are specified):

In a real configuration, either topics or topicsPattern is usually specified.

Hocon YAML

kafka {
    someConsumer {
        topics = ["topic1", "topic2"] //(1)!
        topicsPattern = "topic*" //(2)!
        allowEmptyRecords = false //(3)!
        offset = "latest" //(4)!
        pollTimeout = "5s" //(5)!
        backoffTimeout = "15s" //(6)!
        partitionRefreshInterval = "1m" //(7)!
        threads = 1 //(8)!
        shutdownWait = "30s" //(9)!
        initializationFailTimeout = "30s" //(10)!
        driverProperties { //(11)!
            "bootstrap.servers": "localhost:9093"
            "group.id": "my-group-id"
        }
        telemetry {
            logging {
                enabled = false //(12)!
            }
            metrics {
                enabled = false //(13)!
                driverMetrics = false //(14)!
                slo = [1, 10, 50, 100, 200, 500, 1000, 2000, 5000, 10000, 20000, 30000, 60000, 90000] //(15)!
                tags = { //(16)!
                    "key1" = "value1"
                    "key2" = "value2"
                }
            }
            tracing {
                enabled = true //(17)!
                attributes = { //(18)!
                    "key1" = "value1"
                    "key2" = "value2"
                }
            }
        }
    }
}

1. List of topic values the Consumer subscribes to (not set by default, optional; either topics or topicsPattern must be specified)
2. topic pattern the Consumer subscribes to (not set by default, optional; either topics or topicsPattern must be specified) Supported by the subscribe strategy only; the assign strategy rejects it at startup and requires an explicit topics list.
3. Whether to process empty batches when the signature accepts ConsumerRecords (default: false) If false and ConsumerRecords is empty (no messages), consumer method will not be called. If true, method will be called with empty ConsumerRecords (useful for periodic checks).
4. Initial read position for the assign strategy when group.id is not specified (default: latest). Valid values:
   1. earliest - earliest available offset
   2. latest - latest available offset
   3. string in Duration format, for example 5m, - shift back by the specified duration Format: number + unit (ms, s, m, h, d). Examples: 5m = 5 minutes ago, 1h = 1 hour ago.
5. Maximum time to wait for messages from a topic within one poll() call (default: 5s)
6. Initial delay between unexpected processing errors; with repeated errors the delay doubles up to 60s (default: 15s) If consumer throws unexpected exception (not KafkaSkipRecordException), Kora will restart consumer with backoffTimeout delay to prevent cyclic errors.
7. Partition list refresh period for the assign strategy (default: 1m)
8. Number of threads the consumer starts on; if set to 0, the consumer is not started (default: 1)
9. Time to wait for processing before stopping the consumer during graceful shutdown (default: 30s)
10. Maximum time the container waits during application startup until every consumer thread has completed its first poll() (not set by default, optional) If the timeout expires, application startup fails. When it is not set, the consumer connects in the background and an unavailable broker does not block startup.
11. Official Kafka Consumer Properties; see Apache Kafka Consumer Configs (required, not set by default)
12. Enables module logging (default: false)
13. Enables module metrics (default: false)
14. Registers Apache Kafka driver metrics of the underlying KafkaConsumer in the MeterRegistry (default: false)
15. Configures SLO for metrics (default: io.koraframework.telemetry.common.TelemetryConfig.MetricsConfig#DEFAULT_SLO)
16. Configures metric tags (default: {})
17. Enables module tracing (default: true)
18. Configures tracing attributes (default: {})

kafka:
  someConsumer:
    topics: #(1)!
      - "topic1"
      - "topic2"
    topicsPattern: "topic*" #(2)!
    allowEmptyRecords: false #(3)!
    offset: "latest" #(4)!
    pollTimeout: "5s" #(5)!
    backoffTimeout: "15s" #(6)!
    partitionRefreshInterval: "1m" #(7)!
    threads: 1 #(8)!
    shutdownWait: "30s" #(9)!
    initializationFailTimeout: "30s" #(10)!
    driverProperties: #(11)!
      bootstrap.servers: "localhost:9093"
      group.id: "my-group-id"
    telemetry:
      logging:
        enabled: false #(12)!
      metrics:
        enabled: false #(13)!
        driverMetrics: false #(14)!
        slo: [ 1, 10, 50, 100, 200, 500, 1000, 2000, 5000, 10000, 20000, 30000, 60000, 90000 ] #(15)!
        tags: #(16)!
          key1: value1
          key2: value2
      tracing:
        enabled: true #(17)!
        attributes: #(18)!
          key1: value1
          key2: value2

1. List of topic values the Consumer subscribes to (not set by default, optional; either topics or topicsPattern must be specified)
2. topic pattern the Consumer subscribes to (not set by default, optional; either topics or topicsPattern must be specified) Supported by the subscribe strategy only; the assign strategy rejects it at startup and requires an explicit topics list.
3. Whether to process empty batches when the signature accepts ConsumerRecords (default: false) If false and ConsumerRecords is empty (no messages), consumer method will not be called. If true, method will be called with empty ConsumerRecords (useful for periodic checks).
4. Initial read position for the assign strategy when group.id is not specified (default: latest). Valid values:
   1. earliest - earliest available offset
   2. latest - latest available offset
   3. string in Duration format, for example 5m, - shift back by the specified duration Format: number + unit (ms, s, m, h, d). Examples: 5m = 5 minutes ago, 1h = 1 hour ago.
5. Maximum time to wait for messages from a topic within one poll() call (default: 5s)
6. Initial delay between unexpected processing errors; with repeated errors the delay doubles up to 60s (default: 15s) If consumer throws unexpected exception (not KafkaSkipRecordException), Kora will restart consumer with backoffTimeout delay to prevent cyclic errors.
7. Partition list refresh period for the assign strategy (default: 1m)
8. Number of threads the consumer starts on; if set to 0, the consumer is not started (default: 1)
9. Time to wait for processing before stopping the consumer during graceful shutdown (default: 30s)
10. Maximum time the container waits during application startup until every consumer thread has completed its first poll() (not set by default, optional) If the timeout expires, application startup fails. When it is not set, the consumer connects in the background and an unavailable broker does not block startup.
11. Official Kafka Consumer Properties; see Apache Kafka Consumer Configs (required, not set by default)
12. Enables module logging (default: false)
13. Enables module metrics (default: false)
14. Registers Apache Kafka driver metrics of the underlying KafkaConsumer in the MeterRegistry (default: false)
15. Configures SLO for metrics (default: io.koraframework.telemetry.common.TelemetryConfig.MetricsConfig#DEFAULT_SLO)
16. Configures metric tags (default: {})
17. Enables module tracing (default: true)
18. Configures tracing attributes (default: {})

Module metrics are described in the Metrics Reference section.

Consume strategy

subscribe strategy is used when group.id is present in driverProperties. Application instances then join one consumer group, and Kafka distributes partitions between them so that different instances do not process the same records at the same time.

Example of subscribe strategy configuration:

Hocon YAML

kafka {
    someConsumer {
        topics = ["first"]
        driverProperties {
          "group.id": "my-group-id"
          "bootstrap.servers": "localhost:9093"
        }
    }
}

kafka:
  someConsumer:
    topics:
      - "first"
    driverProperties:
      "group.id": "my-group-id"
      "bootstrap.servers": "localhost:9093"

assign strategy is used when group.id is not specified in driverProperties. Each application instance then assigns partitions of the configured topics to itself, so every instance receives the same records independently. This strategy is useful, for example, when all application replicas must receive the same message at once: to reset a local cache, update local reference data, or handle a service event.

The assign strategy requires an explicit topics list and does not support topicsPattern. The partition list is refreshed every partitionRefreshInterval and split between threads consumers, and the initial read position is controlled by offset.

Example of assign strategy configuration:

Hocon YAML

kafka {
    someConsumer {
        topics = ["first"]
        driverProperties {
          "bootstrap.servers": "localhost:9093"
        }
    }
}

kafka:
  someConsumer:
    topics:
      - "first"
    driverProperties:
      "bootstrap.servers": "localhost:9093"

Signatures

Available signatures for out-of-the-box Kafka Consumer methods, where K refers to the key type and V to the message value type. The generator supports three signature families: separate key/value arguments, a single ConsumerRecord<K, V>, or a whole ConsumerRecords<K, V> batch. These families cannot be mixed in the same method.

Handlers are synchronous: the poll loop thread calls the method and waits for it to return before it polls again. In Kotlin a listener may be declared suspend; the generated handler then runs it through runBlocking on the same poll loop thread, so it still occupies the consumer thread for the whole processing time.

Key and value

A signature with separate arguments accepts value, optional key, optional Headers, optional Consumer<K, V>, and optional deserialization errors. One user argument is treated as value; two user arguments are treated as key and value in that exact order. If key is not declared, the key deserialization type is considered to be byte[].

To handle deserialization errors, add Exception, Throwable, RecordKeyDeserializationException, or RecordValueDeserializationException. When such an argument is present, Kora passes the deserialization error to it, and the corresponding key or value is passed as null. Without such an argument, the deserialization error is thrown from the handler, and the record is read again without committing the current offset.

Java Kotlin

@KafkaListener("kafka.someConsumer")
void process(K key, V value, Headers headers) {
    // some value handling work
}

@KafkaListener("kafka.someConsumer")
fun process(key: K, value: V, headers: Headers) {
    // some value handling work
}

Java Kotlin

@KafkaListener("kafka.someOtherConsumer")
void process(@Nullable V value, @Nullable Exception exception) {
    if (exception != null) {
        // do deserialization handling work
    } else {
        // some value handling work
    }
}

@KafkaListener("kafka.someOtherConsumer")
fun process(value: V?, exception: Exception?) {
    if (exception != null) {
        // do deserialization handling work
    } else {
        // some value handling work
    }
}

Whole record

A signature with ConsumerRecord<K, V> accepts one whole record, optional Consumer<K, V>, and optional deserialization errors: Exception, Throwable, RecordKeyDeserializationException, or RecordValueDeserializationException. Headers and separate key/value arguments are not supported in this signature, because ConsumerRecord already carries them.

If error arguments are not declared, the deserialization error can be thrown when calling record.key() or record.value(). If error arguments are declared, Kora calls key() and/or value() beforehand, catches the deserialization error, and passes it to the method.

Java Kotlin

@KafkaListener("kafka.someConsumer")
void process(ConsumerRecord<K, V> record) {
    try {
        var key = record.key();
        var value = record.value();

        // some value handling work
    } catch (RecordKeyDeserializationException e) {
        // do deserialization handling work
    } catch (RecordValueDeserializationException e) {
        // do deserialization handling work
    }
}

@KafkaListener("kafka.someConsumer")
fun process(record: ConsumerRecord<K, V>) {
    try {
        val key = record.key()
        val value = record.value()

        // some value handling work
    } catch (e: RecordKeyDeserializationException) {
        // do deserialization handling work
    } catch (e: RecordValueDeserializationException) {
        // do deserialization handling work
    }
}

Java Kotlin

@KafkaListener("kafka.someConsumer")
void process(ConsumerRecord<K, V> record,
             @Nullable RecordKeyDeserializationException keyException,
             @Nullable RecordValueDeserializationException valueException) {
    if (keyException != null || valueException != null) {
        // do deserialization handling work
        return;
    }

    var key = record.key();
    var value = record.value();
    // some value handling work
}

@KafkaListener("kafka.someConsumer")
fun process(
    record: ConsumerRecord<K, V>,
    keyException: RecordKeyDeserializationException?,
    valueException: RecordValueDeserializationException?,
) {
    if (keyException != null || valueException != null) {
        // do deserialization handling work
        return
    }

    val key = record.key()
    val value = record.value()
    // some value handling work
}

Batch of records

A signature with ConsumerRecords<K, V> accepts the whole batch of records from one poll(). Together with it, only Consumer<K, V> can be declared. Separate key/value arguments, Headers, and deserialization error arguments are not supported in this signature; deserialization errors should be handled while iterating over records.

Java Kotlin

@KafkaListener("kafka.someConsumer")
void process(ConsumerRecords<K, V> records) {
    for (var record : records) {
        try {
            var key = record.key();
            var value = record.value();

            // some value handling work
        } catch (RecordKeyDeserializationException e) {
            // do deserialization handling work
        } catch (RecordValueDeserializationException e) {
            // do deserialization handling work
        }
    }
}

@KafkaListener("kafka.someConsumer")
fun process(records: ConsumerRecords<K, V>) {
    for (record in records) {
        try {
            val key = record.key()
            val value = record.value()

            // some value handling work
        } catch (e: RecordKeyDeserializationException) {
            // do deserialization handling work
        } catch (e: RecordValueDeserializationException) {
            // do deserialization handling work
        }
    }
}

Offset commit

If the signature does not declare a Consumer<K, V> argument, Kora commits the offset automatically: after each record for key/value and ConsumerRecord<K, V> signatures, or after the whole batch for ConsumerRecords<K, V>.

Automatic commit is only performed when the driver does not commit by itself. If enable.auto.commit is not set in driverProperties, Kora forces it to false and commits offsets itself. If enable.auto.commit is explicitly set to true, the driver owns the offsets and Kora does not commit anything.

If the signature declares a Consumer<K, V> argument, automatic offset commit is disabled, and the handler is fully responsible for calling commitSync() or commitAsync(). This mode is useful when the offset should be committed only after an external operation, several records should be committed together, or the read position should be controlled manually.

In subscribe mode, a manual commit commits the offset inside the consumer group. In assign mode, partitions are not coordinated through a consumer group, so it is usually more important to manage the position manually with seek(), pause(), and resume() instead of relying on a group offset commit. If the handler fails before the manual commit, the record or batch will be read again according to the current consumer position.

Java Kotlin

@KafkaListener("kafka.someConsumer")
void process(ConsumerRecord<K, V> record, Consumer<K, V> consumer) {
    try {
        var key = record.key();
        var value = record.value();

        // some value handling work
    } catch (RecordKeyDeserializationException e) {
        // do deserialization handling work
    } catch (RecordValueDeserializationException e) {
        // do deserialization handling work
    } finally {
        consumer.commitSync();
    }
}

@KafkaListener("kafka.someConsumer")
fun process(record: ConsumerRecord<K, V>, consumer: Consumer<K, V>) {
    try {
        val key = record.key()
        val value = record.value()

        // some value handling work
    } catch (e: RecordKeyDeserializationException) {
        // do deserialization handling work
    } catch (e: RecordValueDeserializationException) {
        // do deserialization handling work
    } finally {
        consumer.commitSync()
    }
}

Deserialization

Deserializer is used to deserialize ConsumerRecord keys and values. Kora provides Deserializer components for basic types: String, UUID, byte[], Bytes, ByteBuffer, Double, Float, Integer, Long, Short, and Void.

Tags are supported to better customize the Deserializer. Tags can be set on parameter-key, parameter-value, as well as on parameters of type ConsumerRecord and ConsumerRecords. These tags will be set on container dependencies.

Java Kotlin

@Component
final class ConsumerService {

    @KafkaListener("kafka.someConsumer1")
    void process1(@Tag(Sometag1.class) String key, @Tag(Sometag2.class) String value) {
        // some handler code
    }

    @KafkaListener("kafka.someConsumer2")
    void process2(ConsumerRecord<@Tag(Sometag1.class) String, @Tag(Sometag2.class) String> record) {
        // some handler code
    }
}

@Component
class ConsumerService {
    @KafkaListener("kafka.someConsumer1")
    fun process1(@Tag(Sometag1::class) key: String, @Tag(Sometag2::class) value: String) {
        // some handler code
    }

    @KafkaListener("kafka.someConsumer2")
    fun process2(record: ConsumerRecord<@Tag(Sometag1::class) String, @Tag(Sometag2::class) String>) {
        // some handler code
    }
}

If deserialization from JSON is required, use the @Json tag. In this case, Kora uses JsonReader<T> and JsonKafkaDeserializer<T> from the JSON module:

Java Kotlin

@Component
final class ConsumerService {

    @Json
    public record JsonEvent(String name, Integer code) {}

    @KafkaListener("kafka.someConsumer1")
    void process1(String key, @Json JsonEvent value) {
        // some handler code
    }

    @KafkaListener("kafka.someConsumer2")
    void process2(ConsumerRecord<String, @Json JsonEvent> record) {
        // some handler code
    }
}

@Component
class ConsumerService {

    @Json
    data class JsonEvent(val name: String, val code: Int)

    @KafkaListener("kafka.someConsumer1")
    fun process1(key: String, @Json value: JsonEvent) {
        // some handler code
    }

    @KafkaListener("kafka.someConsumer2")
    fun process2(record: ConsumerRecord<String, @Json JsonEvent>) {
        // some handler code
    }
}

For non-key handlers, the default is Deserializer<byte[]> since it simply returns unhandled bytes.

Custom Deserializer

If custom deserialization is required, you can implement your own Deserializer.

Option 1: Default deserializer for type

If you provide Deserializer<T> as a component without a tag, it will be used for all consumers of this type:

Java Kotlin

@Component
public static class MyEventDeserializer implements Deserializer<MyEvent> {

    private final JsonReader<MyEvent> reader;

    public MyEventDeserializer(JsonReader<MyEvent> reader) {
        this.reader = reader;
    }

    @Override
    public MyEvent deserialize(String topic, byte[] data) {
        return reader.read(data);
    }
}

@Component
final class SomeConsumer {

    @KafkaListener("kafka.someConsumer")
    void process(MyEvent value) { // Uses MyEventDeserializer
        // event handling
    }
}

@Component
class MyEventDeserializer(
    private val reader: JsonReader<MyEvent>
) : Deserializer<MyEvent> {

    override fun deserialize(topic: String, data: ByteArray): MyEvent {
        return requireNotNull(reader.read(data)) { "Empty payload in topic $topic" }
    }
}

@Component
class SomeConsumer {

    @KafkaListener("kafka.someConsumer")
    fun process(value: MyEvent) { // Uses MyEventDeserializer
        // event handling
    }
}

JsonReader<T>.read(byte[]) throws an unchecked JacksonException on malformed payload and returns null for a null payload, so in Kotlin the result must be unwrapped with requireNotNull before it is returned as a non-null type.

Option 2: Point deserializer for specific consumer

If you need to use different deserialization for different consumers of the same type, you can use tags:

Java Kotlin

@Component
final class SomeConsumer {

    @Json
    public record MyEvent(String username, int code) {}

    @Tag(MyEvent.class)
    @Component
    public static class MyDeserializer implements Deserializer<MyEvent> {

        private final JsonReader<MyEvent> reader;

        public MyDeserializer(JsonReader<MyEvent> reader) {
            this.reader = reader;
        }

        @Override
        public MyEvent deserialize(String topic, byte[] data) {
            return reader.read(data);
        }
    }

    @KafkaListener("kafka.someConsumer")
    void process(@Tag(MyEvent.class) MyEvent value) {
        // event handling
    }
}

@Component
class SomeConsumer {

    @Json
    data class MyEvent(val username: String, val code: Int)

    @Tag(MyEvent::class)
    @Component
    class MyDeserializer(
        private val reader: JsonReader<MyEvent>
    ) : Deserializer<MyEvent> {

        override fun deserialize(topic: String, data: ByteArray): MyEvent {
            return requireNotNull(reader.read(data)) { "Empty payload in topic $topic" }
        }
    }

    @KafkaListener("kafka.someConsumer")
    fun process(@Tag(MyEvent::class) value: MyEvent) {
        // event handling
    }
}

Exception handling

If the method labeled @KafkaListener throws an exception, the poll loop is interrupted and the Consumer is restarted after backoffTimeout, because there is no general solution on how to handle this and the developer must decide how to handle it. With repeated errors the delay doubles up to 60s, so a permanently failing handler does not spin the broker.

Exception skipping

If you need to skip processing a specific event (ConsumerRecord) during processing for business logic reasons, you can throw a KafkaSkipRecordException by passing the actual exception to the constructor. In this case, all metrics will be correctly accounted for and recorded, processing of the corresponding event will be skipped, and the next event will begin to be processed.

Java Kotlin

@Component
final class SomeConsumer {

    @KafkaListener("kafka.someConsumer1")
    void process1(String key, String value) {
        if ("skip".equals(value)) {
            throw new KafkaSkipRecordException(new IllegalArgumentException("Want to skip!"));
        }
        // some handler code
    }
}

@Component
class SomeConsumer {

    @KafkaListener("kafka.someConsumer1")
    fun process1(key: String, value: String) {
        if (value == "skip") {
            throw KafkaSkipRecordException(IllegalArgumentException("Want to skip!"))
        }
        // some handler code
    }
}

If you want to implement your own skippable exceptions, you can use the SkippableRecordException interface, which should be implemented in your exceptions.

Java Kotlin

public class MyKafkaSkipRecordException extends RuntimeException implements SkippableRecordException {

}

class MyKafkaSkipRecordException : RuntimeException(), SkippableRecordException

Skipping applies to the single-record signatures only: the batch signature receives the whole ConsumerRecords and decides itself which records to skip.

Deserialization errors

If you use a signature with ConsumerRecord or ConsumerRecords, you will get a value deserialization exception at the moment of calling the key or value methods. At that point, it is worth handling it in the way you want.

The following exceptions are thrown:

• io.koraframework.kafka.common.exceptions.RecordKeyDeserializationException.
• io.koraframework.kafka.common.exceptions.RecordValueDeserializationException.

Both extend org.apache.kafka.common.errors.SerializationException. From these exceptions, you can get a raw ConsumerRecord<byte[], byte[]> using getRecord() method:

Java Kotlin

@Component
final class ConsumerService {

    @KafkaListener("kafka.someConsumer")
    public void process(ConsumerRecord<String, String> record) {
        try {
            var key = record.key();
            var value = record.value();
            // some value handling work
        } catch (RecordKeyDeserializationException e) {
            ConsumerRecord<byte[], byte[]> rawRecord = e.getRecord();
            // Handle raw record (log, send to DLQ, etc.)
        } catch (RecordValueDeserializationException e) {
            ConsumerRecord<byte[], byte[]> rawRecord = e.getRecord();
            // Handle raw record (log, send to DLQ, etc.)
        }
    }
}

@Component
class ConsumerService {

    @KafkaListener("kafka.someConsumer")
    fun process(record: ConsumerRecord<String, String>) {
        try {
            val key = record.key()
            val value = record.value()
            // some value handling work
        } catch (e: RecordKeyDeserializationException) {
            val rawRecord = e.record
            // Handle raw record (log, send to DLQ, etc.)
        } catch (e: RecordValueDeserializationException) {
            val rawRecord = e.record
            // Handle raw record (log, send to DLQ, etc.)
        }
    }
}

If you use a signature with unpacked key/value/headers, you can add Exception, Throwable, RecordKeyDeserializationException or RecordValueDeserializationException as the last argument.

Java Kotlin

@Component
final class ConsumerService {

    @KafkaListener("kafka.someConsumer")
    public void process(@Nullable String key, @Nullable String value, @Nullable Exception exception) {
        if (exception != null) {
            // handle exception
        } else {
            // handle key/value
        }
    }
}

@Component
class ConsumerService {

    @KafkaListener("kafka.someConsumer")
    fun process(key: String?, value: String?, exception: Exception?) {
        if (exception != null) {
            // handle exception
        } else {
            // handle key/value
        }
    }
}

Note that all arguments become optional, meaning we expect to either have a key and value or an exception. When both a key and a value fail to deserialize and a single Exception argument is declared, the key error is passed.

Custom tag

Automatic tag is created for the consumer by default; it is named <ListenerClass>Module.<ListenerClass><Method>Tag and can be viewed in the generated module at compile time.

If for some reason you need to override the consumer tag, you can set it as an argument to the @KafkaListener annotation:

Java Kotlin

@Component
final class ConsumerService {

    @KafkaListener(value = "kafka.someConsumer", tag = ConsumerService.class)
    public void process(String value) {

    }
}

@Component
class ConsumerService {

    @KafkaListener(value = "kafka.someConsumer", tag = ConsumerService::class)
    fun process(value: String) {

    }
}

The tag is applied to the generated KafkaListenerConfig, the handler, and the rebalance listener dependency of the container, so it is exactly the tag a ConsumerAwareRebalanceListener must be registered under.

Rebalance events

You can listen and react to rebalance events with your implementation of the ConsumerAwareRebalanceListener interface, it should be provided as a component by the consumer tag:

Java Kotlin

@Tag(SomeListenerProcessTag.class)
@Component
public final class SomeListener implements ConsumerAwareRebalanceListener {

    @Override
    public void onPartitionsRevoked(Consumer<?, ?> consumer, Collection<TopicPartition> partitions) {
        // Called before partitions are revoked from this consumer.
        // Use this to commit offsets or cleanup state.
    }

    @Override
    public void onPartitionsAssigned(Consumer<?, ?> consumer, Collection<TopicPartition> partitions) {
        // Called when partitions are assigned to this consumer.
        // Use this to initialize state for assigned partitions.
    }

    @Override
    public void onPartitionsLost(Consumer<?, ?> consumer, Collection<TopicPartition> partitions) {
        // Called when partitions are lost (e.g., consumer failure, group rebalance).
        // Unlike onPartitionsRevoked, this is called when the consumer is no longer
        // part of the group and cannot commit offsets.
        // Use this to cleanup local state for lost partitions.
    }
}

@Tag(SomeListenerProcessTag::class)
@Component
class SomeListener : ConsumerAwareRebalanceListener {

    override fun onPartitionsRevoked(consumer: Consumer<*, *>, partitions: Collection<TopicPartition>) {
        // Called before partitions are revoked from this consumer.
        // Use this to commit offsets or cleanup state.
    }

    override fun onPartitionsAssigned(consumer: Consumer<*, *>, partitions: Collection<TopicPartition>) {
        // Called when partitions are assigned to this consumer.
        // Use this to initialize state for assigned partitions.
    }

    override fun onPartitionsLost(consumer: Consumer<*, *>, partitions: Collection<TopicPartition>) {
        // Called when partitions are lost (e.g., consumer failure, group rebalance).
        // Unlike onPartitionsRevoked, this is called when the consumer is no longer
        // part of the group and cannot commit offsets.
        // Use this to cleanup local state for lost partitions.
    }
}

onPartitionsLost has a default implementation that delegates to onPartitionsRevoked, so it only has to be overridden when lost partitions require different handling than revoked ones.

Rebalance events exist only in the subscribe strategy: the assign container manages partitions itself and never consults the listener.

Manual override

Kora provides a small wrapper over KafkaConsumer that allows you to easily trigger the handling of incoming events. Both containers implement GeneratedListener, so they take part in the application graph lifecycle like any generated container.

The subscribe container constructor is as follows:

public KafkaSubscribeConsumerContainer(String listenerConfig,
                                       String listenerImpl,
                                       KafkaListenerConfig config,
                                       Deserializer<K> keyDeserializer,
                                       Deserializer<V> valueDeserializer,
                                       BaseKafkaRecordsHandler<K, V> handler,
                                       KafkaConsumerTelemetry telemetry,
                                       @Nullable ConsumerAwareRebalanceListener rebalanceListener)

The assign container constructor is as follows:

public KafkaAssignConsumerContainer(String listenerConfig,
                                    String listenerImpl,
                                    KafkaListenerConfig config,
                                    Deserializer<K> keyDeserializer,
                                    Deserializer<V> valueDeserializer,
                                    KafkaConsumerTelemetry telemetry,
                                    BaseKafkaRecordsHandler<K, V> handler)

listenerConfig is the configuration path used in telemetry, and listenerImpl is the logger name of the listener.

BaseKafkaRecordsHandler<K,V> is the basic functional interface of the handler:

@FunctionalInterface
public interface BaseKafkaRecordsHandler<K, V> {

    void handle(KafkaConsumerPollObservation observation,
                ConsumerRecords<K, V> records,
                Consumer<K, V> consumer,
                boolean commitAllowed);
}

commitAllowed reports whether the driver leaves offset management to the handler, that is whether enable.auto.commit is disabled. Ready-made wrappers for per-record and per-batch handling are available in HandlerWrapper.

Telemetry

Kafka uses a telemetry contract for logging, metrics, and tracing of messages. Telemetry configuration (the telemetry { logging / metrics / tracing } section) is described in the Configuration section.

KafkaConsumerTelemetryFactory builds a KafkaConsumerTelemetry for each listener out of the listener configuration path, the listener class name, driverProperties, and KafkaConsumerTelemetryConfig. KafkaConsumerTelemetry opens a KafkaConsumerPollObservation for every poll() call, derives a KafkaConsumerRecordObservation for every record, and reports the consumer lag per partition:

public interface KafkaConsumerTelemetry {

    MeterRegistry meterRegistry();

    KafkaConsumerPollObservation observePoll();

    void reportLag(TopicPartition partition, long lag);
}

Every observation is closed when processing completes, and the record observation carries the topic, partition, offset, and processing duration.

The default implementation is DefaultKafkaConsumerTelemetryFactory, registered by KafkaModule as a @DefaultComponent. It disables itself entirely when logging, metrics, and tracing are all off, and otherwise builds the enabled parts out of:

• DefaultKafkaConsumerLoggerFactory builds the logger that records the start and end of polling and message processing;
• DefaultKafkaConsumerMetricsFactory builds the meters for batch duration, record duration, and lag.

Both factories are injected into KafkaModule as optional dependencies, so providing your own @Component subclass of either one replaces just that part of the default telemetry. Providing your own KafkaConsumerTelemetryFactory component replaces telemetry entirely.

Metrics and tracing are described in the Metrics Reference section.

Producer

Producer sends records to a topic. Kora creates an implementation of the interface annotated with @KafkaPublisher, selects a Serializer for the key and value, calls KafkaProducer#send, and connects sending with telemetry.

To create a Producer, use the @KafkaPublisher annotation on an interface. To send messages to an arbitrary topic, declare a method with a ProducerRecord parameter:

Java Kotlin

@KafkaPublisher("kafka.someProducer")
public interface MyPublisher {
      void send(ProducerRecord<String, String> record);
}

@KafkaPublisher("kafka.someProducer")
interface MyPublisher {
    fun send(record: ProducerRecord<String, String>)
}

The annotation parameter indicates the path to the producer configuration.

Topic

If typed methods are required for specific topic values, use the @KafkaPublisher.Topic annotation:

Java Kotlin

@KafkaPublisher("kafka.someProducer")
public interface MyPublisher {

    @KafkaPublisher.Topic("kafka.someProducer.someTopic")
    void send(String value);
}

@KafkaPublisher("kafka.someProducer")
interface MyPublisher {

    @KafkaPublisher.Topic("kafka.someProducer.someTopic")
    fun send(value: String)
}

The annotation parameter indicates the path for the topic configuration. A path that starts with . is resolved relative to the @KafkaPublisher configuration path, so @KafkaPublisher.Topic(".someTopic") on a publisher configured at kafka.someProducer reads kafka.someProducer.someTopic.

Configuration

Configuration describes the settings of a particular @KafkaPublisher; below is an example for the kafka.someProducer configuration path.

Basic configuration parameters:

Hocon YAML

kafka {
    someProducer {
        driverProperties { //(1)!
          "bootstrap.servers": "localhost:9093"
        }
    }
}

1. Official Kafka Producer Properties (required, no default)

kafka:
  someProducer:
    driverProperties: #(1)!
      "bootstrap.servers": "localhost:9093"

1. Official Kafka Producer Properties (required, no default)

Full Configuration

Example of the complete configuration described in the KafkaPublisherConfig class (default or example values are specified):

Hocon YAML

kafka {
    someProducer {
        driverProperties { //(1)!
          "bootstrap.servers": "localhost:9093"
        }
        telemetry {
          logging {
            enabled = false //(2)!
          }
          metrics {
            enabled = false //(3)!
            driverMetrics = false //(4)!
            slo = [ 1, 10, 50, 100, 200, 500, 1000, 2000, 5000, 10000, 20000, 30000, 60000, 90000 ] //(5)!
            tags = { //(6)!
              "key1" = "value1"
              "key2" = "value2"
            }
          }
          tracing {
            enabled = true //(7)!
            attributes = { //(8)!
              "key1" = "value1"
              "key2" = "value2"
            }
          }
        }
    }
}

1. Official Kafka Producer Properties; see Apache Kafka Producer Configs (required, not set by default)
2. Enables module logging (default: false)
3. Enables module metrics (default: false)
4. Registers Apache Kafka driver metrics of the underlying KafkaProducer in the MeterRegistry (default: false)
5. Configures SLO for metrics (default: io.koraframework.telemetry.common.TelemetryConfig.MetricsConfig#DEFAULT_SLO)
6. Configures metric tags (default: {})
7. Enables module tracing (default: true)
8. Configures tracing attributes (default: {})

kafka:
  someProducer:
    driverProperties: #(1)!
      bootstrap.servers: "localhost:9093"
    telemetry:
      logging:
        enabled: false #(2)!
      metrics:
        enabled: false #(3)!
        driverMetrics: false #(4)!
        slo: [ 1, 10, 50, 100, 200, 500, 1000, 2000, 5000, 10000, 20000, 30000, 60000, 90000 ] #(5)!
        tags: #(6)!
          key1: value1
          key2: value2
      tracing:
        enabled: true #(7)!
        attributes: #(8)!
          key1: value1
          key2: value2

1. Official Kafka Producer Properties; see Apache Kafka Producer Configs (required, not set by default)
2. Enables module logging (default: false)
3. Enables module metrics (default: false)
4. Registers Apache Kafka driver metrics of the underlying KafkaProducer in the MeterRegistry (default: false)
5. Configures SLO for metrics (default: io.koraframework.telemetry.common.TelemetryConfig.MetricsConfig#DEFAULT_SLO)
6. Configures metric tags (default: {})
7. Enables module tracing (default: true)
8. Configures tracing attributes (default: {})

topic configuration describes the settings of a particular @KafkaPublisher.Topic; below is an example for the kafka.someProducer.someTopic configuration path.

Example of the complete configuration described in the KafkaPublisherConfig.TopicConfig class (default or example values are specified):

Hocon YAML

kafka {
  someProducer {
    someTopic {
      topic = "my-topic" //(1)!
      partition = 1 //(2)!
    }
  }
}

1. topic where the method sends data (required, not set by default)
2. topic partition where the method sends data (not set by default, optional) If specified, all messages will be sent to the specified partition. If not specified, standard Kafka partitioning is used (by key or random).

kafka:
  someProducer:
    someTopic:
      topic: "my-topic" #(1)!
      partition: 1 #(2)!

1. topic where the method sends data (required, not set by default)
2. topic partition where the method sends data (not set by default, optional) If specified, all messages will be sent to the specified partition. If not specified, standard Kafka partitioning is used (by key or random).

Signatures

Available signatures for out-of-the-box Kafka Producer methods, where K refers to the key type and V to the message value type. The generator supports two signature families: sending a ready ProducerRecord<K, V> and sending through a method annotated with @KafkaPublisher.Topic. These families cannot be mixed in the same method.

Prepared event

A method with ProducerRecord<K, V> is used when the topic, partition, timestamp, or Headers should be set by the calling code. Such a method cannot be annotated with @KafkaPublisher.Topic, because all send details are already contained in the ProducerRecord. One Callback can be passed additionally.

Java Kotlin

@KafkaPublisher("kafka.someProducer")
public interface MyPublisher {

    void send(ProducerRecord<K, V> record);

    void send(ProducerRecord<K, V> record, Callback callback);
}

@KafkaPublisher("kafka.someProducer")
interface MyPublisher {

    fun send(record: ProducerRecord<K, V>)

    fun send(record: ProducerRecord<K, V>, callback: Callback)
}

Methods per topic

A method with key, value, and Headers must be annotated with @KafkaPublisher.Topic. One user argument is treated as value; two user arguments are treated as key and value in that exact order. Headers and Callback can be declared additionally, but only one argument of each type is allowed. If Headers is not passed, Kora creates empty headers.

Java Kotlin

@KafkaPublisher("kafka.someProducer")
public interface MyPublisher {

    @KafkaPublisher.Topic("kafka.someProducer.someTopic")
    void send(V value);

    @KafkaPublisher.Topic("kafka.someProducer.someTopic")
    void send(K key, V value);

    @KafkaPublisher.Topic("kafka.someProducer.someTopic")
    void send(K key, V value, Headers headers);

    @KafkaPublisher.Topic("kafka.someProducer.someTopic")
    void send(K key, V value, Headers headers, Callback callback);
}

@KafkaPublisher("kafka.someProducer")
interface MyPublisher {

    @KafkaPublisher.Topic("kafka.someProducer.someTopic")
    fun send(value: V)

    @KafkaPublisher.Topic("kafka.someProducer.someTopic")
    fun send(key: K, value: V)

    @KafkaPublisher.Topic("kafka.someProducer.someTopic")
    fun send(key: K, value: V, headers: Headers)

    @KafkaPublisher.Topic("kafka.someProducer.someTopic")
    fun send(key: K, value: V, headers: Headers, callback: Callback)
}

Send result

For a synchronous method, the return type can be void/Unit or RecordMetadata. In this case, Kora calls KafkaProducer#send, waits for send completion through Future#get(), and only then returns control to the caller.

For asynchronous sending, the return type can be Future<RecordMetadata>, CompletionStage<RecordMetadata>, or CompletableFuture<RecordMetadata>. In Kotlin, suspend methods and Deferred<RecordMetadata> are also supported. If the signature contains a Callback, Kora first completes its own send telemetry and then calls the user Callback.

Java Kotlin

@KafkaPublisher("kafka.someProducer")
public interface MyPublisher {

    @KafkaPublisher.Topic("kafka.someProducer.someTopic")
    RecordMetadata send(V value);

    @KafkaPublisher.Topic("kafka.someProducer.someTopic")
    Future<RecordMetadata> sendFuture(V value);

    @KafkaPublisher.Topic("kafka.someProducer.someTopic")
    CompletionStage<RecordMetadata> sendStage(V value);

    @KafkaPublisher.Topic("kafka.someProducer.someTopic")
    CompletableFuture<RecordMetadata> sendCompletableFuture(V value);
}

@KafkaPublisher("kafka.someProducer")
interface MyPublisher {

    @KafkaPublisher.Topic("kafka.someProducer.someTopic")
    fun send(value: V): RecordMetadata

    @KafkaPublisher.Topic("kafka.someProducer.someTopic")
    suspend fun sendSuspend(value: V): RecordMetadata

    @KafkaPublisher.Topic("kafka.someProducer.someTopic")
    fun sendFuture(value: V): Future<RecordMetadata>

    @KafkaPublisher.Topic("kafka.someProducer.someTopic")
    fun sendStage(value: V): CompletionStage<RecordMetadata>

    @KafkaPublisher.Topic("kafka.someProducer.someTopic")
    fun sendCompletableFuture(value: V): CompletableFuture<RecordMetadata>

    @KafkaPublisher.Topic("kafka.someProducer.someTopic")
    fun sendDeferred(value: V): Deferred<RecordMetadata>
}

Invalid combinations are: ProducerRecord<K, V> together with @KafkaPublisher.Topic, ProducerRecord<K, V> together with separate key/value/Headers, more than one Headers, more than one Callback, and a method with separate key/value without @KafkaPublisher.Topic.

Serialization

Serializer is used to serialize ProducerRecord keys and values. Kora provides Serializer components for basic types: String, UUID, byte[], Bytes, ByteBuffer, Double, Float, Integer, Long, Short, and Void.

To specify which Serializer to take from the container, tags can be used. Tags should be set on ProducerRecord or key/value parameters of methods:

Java Kotlin

@KafkaPublisher("kafka.someProducer")
public interface MyKafkaProducer {

    void send(ProducerRecord<@Tag(MyTag1.class) String, @Tag(MyTag2.class) String> record);

    @KafkaPublisher.Topic("kafka.someProducer.someTopic")
    void send(@Tag(MyTag1.class) String key, @Tag(MyTag2.class) String value);
}

@KafkaPublisher("kafka.someProducer")
interface MyKafkaProducer {

    fun send(record: ProducerRecord<@Tag(MyTag1::class) String, @Tag(MyTag2::class) String>)

    @KafkaPublisher.Topic("kafka.someProducer.someTopic")
    fun send(@Tag(MyTag1::class) key: String, @Tag(MyTag2::class) value: String)
}

If serialization to JSON is required, use the @Json tag. In this case, Kora uses JsonWriter<T> and JsonKafkaSerializer<T> from the JSON module:

Java Kotlin

@KafkaPublisher("kafka.someProducer")
public interface MyKafkaProducer {

    @Json
    record JsonEvent(String name, Integer code) {}

    void send(ProducerRecord<String, @Json JsonEvent> record);

    @KafkaPublisher.Topic("kafka.someProducer.someTopic")
    void send(String key, @Json JsonEvent value);
}

@KafkaPublisher("kafka.someProducer")
interface MyKafkaProducer {

    @Json
    data class JsonEvent(val name: String, val code: Int)

    fun send(record: ProducerRecord<String, @Json JsonEvent>)

    @KafkaPublisher.Topic("kafka.someProducer.someTopic")
    fun send(key: String, @Json value: JsonEvent)
}

Default Serializers and Deserializers

KafkaModule automatically provides serializers and deserializers for base types via KafkaSerializersModule and KafkaDeserializersModule.

These serializers/deserializers are provided as @DefaultComponent components without tags and are used by default for all consumers/producers of corresponding types. Because they are default components, providing your own untagged Serializer<T>/Deserializer<T> for the same type overrides them without any conflict.

Supported types out of the box:

Type	Serializer	Deserializer
String	StringSerializer	StringDeserializer
byte[]	ByteArraySerializer	ByteArrayDeserializer
ByteBuffer	ByteBufferSerializer	ByteBufferDeserializer
Bytes	BytesSerializer	BytesDeserializer
UUID	UUIDSerializer	UUIDDeserializer
Integer	IntegerSerializer	IntegerDeserializer
Long	LongSerializer	LongDeserializer
Short	ShortSerializer	ShortDeserializer
Double	DoubleSerializer	DoubleDeserializer
Float	FloatSerializer	FloatDeserializer
Void	VoidSerializer	VoidDeserializer

Additionally, JsonKafkaSerializer<T> and JsonKafkaDeserializer<T> are provided under the @Json tag for any type that has a generated JsonWriter<T>/JsonReader<T>.

To use, simply specify the type in the publisher/consumer method:

Java Kotlin

@KafkaPublisher("kafka.someProducer")
public interface MyPublisher {
    @KafkaPublisher.Topic("kafka.someProducer.topic")
    void send(UUID key, String value);
}

@KafkaPublisher("kafka.someProducer")
interface MyPublisher {
    @KafkaPublisher.Topic("kafka.someProducer.topic")
    fun send(key: UUID, value: String)
}

Custom Serializer

If custom serialization is required, you can implement your own Serializer.

Option 1: Default serializer for type

If you provide Serializer<T> as a component without a tag, it will be used for all producers of this type:

Java Kotlin

@Component
public static class MyEventSerializer implements Serializer<MyEvent> {

    private final JsonWriter<MyEvent> writer;

    public MyEventSerializer(JsonWriter<MyEvent> writer) {
        this.writer = writer;
    }

    @Override
    public byte[] serialize(String topic, MyEvent data) {
        return writer.toByteArray(data);
    }
}

@KafkaPublisher("kafka.someProducer")
public interface MyPublisher {

    @KafkaPublisher.Topic("kafka.someProducer.topic")
    void send(MyEvent value); // Uses MyEventSerializer
}

@Component
class MyEventSerializer(
    private val writer: JsonWriter<MyEvent>
) : Serializer<MyEvent> {

    override fun serialize(topic: String, data: MyEvent): ByteArray {
        return writer.toByteArray(data)
    }
}

@KafkaPublisher("kafka.someProducer")
interface MyPublisher {

    @KafkaPublisher.Topic("kafka.someProducer.topic")
    fun send(value: MyEvent) // Uses MyEventSerializer
}

JsonWriter<T>.toByteArray(T) throws an unchecked JacksonException, so no checked exception has to be handled in the mapper.

Option 2: Point serializer for specific producer

If you need to use different serialization for different producers of the same type, you can use tags:

Java Kotlin

@KafkaPublisher("kafka.someProducer")
public interface MyKafkaProducer {

    @Json
    record MyEvent(String username, int code) {}

    @Tag(MyEvent.class)
    @Component
    class MySerializer implements Serializer<MyEvent> {

        private final JsonWriter<MyEvent> writer;

        public MySerializer(JsonWriter<MyEvent> writer) {
            this.writer = writer;
        }

        @Override
        public byte[] serialize(String topic, MyEvent data) {
            return writer.toByteArray(data);
        }
    }

    void send(ProducerRecord<String, @Tag(MyEvent.class) MyEvent> record);
}

@KafkaPublisher("kafka.someProducer")
interface MyKafkaProducer {

    @Json
    data class MyEvent(val username: String, val code: Int)

    @Tag(MyEvent::class)
    @Component
    class MySerializer(
        private val writer: JsonWriter<MyEvent>
    ) : Serializer<MyEvent> {

        override fun serialize(topic: String, data: MyEvent): ByteArray {
            return writer.toByteArray(data)
        }
    }

    fun send(record: ProducerRecord<String, @Tag(MyEvent::class) MyEvent>)
}

Exception handling

If a send error happens in a method that returns void/Unit or RecordMetadata, io.koraframework.kafka.common.exceptions.KafkaPublishException is thrown. It extends org.apache.kafka.common.KafkaException, and the original error from KafkaProducer is available in getCause(). A RuntimeException reported by the driver is rethrown as is, without being wrapped.

Methods that return Future<RecordMetadata>, CompletionStage<RecordMetadata>, or CompletableFuture<RecordMetadata> do not throw: the send error completes the returned future exceptionally instead.

Java Kotlin

@Component
class SomeService {

    private final MyPublisher publisher;

    public SomeService(MyPublisher publisher) {
        this.publisher = publisher;
    }

    void sendMessage() {
        try {
            publisher.send("key", "value");
        } catch (KafkaPublishException e) {
            // Handle the failed send (log, retry, etc.)
            var cause = e.getCause();
        }
    }
}

@Component
class SomeService(
    private val publisher: MyPublisher
) {

    fun sendMessage() {
        try {
            publisher.send("key", "value")
        } catch (e: KafkaPublishException) {
            // Handle the failed send (log, retry, etc.)
            val cause = e.cause
        }
    }
}

Serialization errors

If a key or value serialization error happens in a method annotated with @KafkaPublisher.Topic, org.apache.kafka.common.errors.SerializationException is thrown, just like with a direct org.apache.kafka.clients.producer.Producer#send call. Serialization runs before the record is handed to the driver, so such an error is not wrapped into KafkaPublishException.

Transactions

Messages can be sent to Kafka within a transaction. For this, use the @KafkaPublisher annotation and extend TransactionalPublisher.

First, describe a regular KafkaProducer, and then use its type to create a transactional Producer:

Java Kotlin

@KafkaPublisher("kafka.someProducer")
public interface MyPublisher {

    @KafkaPublisher.Topic("kafka.someProducer.someTopic")
    void send(String key, String value);
}

@KafkaPublisher("kafka.someTransactionalProducer")
public interface MyTransactionalPublisher extends TransactionalPublisher<MyPublisher> {

}

@KafkaPublisher("kafka.someProducer")
interface MyPublisher {

    @KafkaPublisher.Topic("kafka.someProducer.someTopic")
    fun send(key: String, value: String)
}


@KafkaPublisher("kafka.someTransactionalProducer")
interface MyTransactionalPublisher : TransactionalPublisher<MyPublisher>

The transactional publisher reuses driverProperties of the delegate publisher and only overrides transactional.id, so the broker connection is configured once, at the delegate publisher configuration path.

Use inTx methods to send messages in a transaction: all messages inside the lambda are committed on successful execution and aborted on error.

Java Kotlin

transactionalPublisher.inTx(publisher -> {
    publisher.send("key1", "value1");
    publisher.send("key2", "value2");
});

transactionalPublisher.inTx(TransactionalConsumer<MyPublisher, RuntimeException> { publisher ->
    publisher.send("key1", "value1")
    publisher.send("key2", "value2")
})

In Kotlin, inTx and withTx are overloaded for a value-returning and a void callback, so the lambda has to be wrapped into an explicitly typed SAM constructor for the compiler to choose the overload.

It is also possible to manage the transaction manually through begin():

Java Kotlin

// commit will be called on try-with-resources close
try (var transaction = transactionalPublisher.begin()) {
    transaction.publisher().send("key1", "value1");
    if (somethingBad) {
        transaction.abort();
    }
}

// commit will be called on use close
transactionalPublisher.begin().use {
    it.publisher().send("key1", "value1")
    if (somethingBad) {
        it.abort()
    }
}

Configuration

KafkaPublisherConfig.TransactionConfig is used to configure @KafkaPublisher with the TransactionalPublisher interface:

Hocon YAML

kafka {
    someTransactionalProducer {
        idPrefix = "kora-app-" //(1)!
        maxPoolSize = 10 //(2)!
        maxWaitTime = "10s" //(3)!
    }
}

1. Transaction identifier prefix; a random UUID will be appended to it (default: kora-app-) Format: {idPrefix}-{uuid}. Example: my-transaction-550e8400-e29b-41d4-a716-446655440000.
2. Maximum size of the transactional Producer pool (default: 10)
3. Maximum time to wait for a free Producer from the pool (default: 10s)

kafka:
  someTransactionalProducer:
    idPrefix: "kora-app-" #(1)!
    maxPoolSize: 10 #(2)!
    maxWaitTime: "10s" #(3)!

1. Transaction identifier prefix; a random UUID will be appended to it (default: kora-app-) Format: {idPrefix}-{uuid}. Example: my-transaction-550e8400-e29b-41d4-a716-446655440000.
2. Maximum size of the transactional Producer pool (default: 10)
3. Maximum time to wait for a free Producer from the pool (default: 10s)

When the pool is exhausted, begin() waits up to maxWaitTime for a free Producer and then throws org.apache.kafka.common.errors.TimeoutException.

Advanced Transaction Usage

Transaction Interface

The begin() method returns a Transaction<P> object that provides advanced transaction management capabilities:

Java Kotlin

try (var tx = transactionalPublisher.begin()) {
    // Sending messages
    tx.publisher().send("key1", "value1");
    tx.publisher().send("key2", "value2");

    // Commit consumer offsets within the transaction (exactly-once semantics)
    Map<TopicPartition, OffsetAndMetadata> offsets = ...;
    ConsumerGroupMetadata groupMetadata = ...;
    tx.sendOffsetsToTransaction(offsets, groupMetadata);

    // Explicit flush to guarantee sending before commit
    tx.flush();

    // commit() is called automatically on try-with-resources close
}

transactionalPublisher.begin().use { tx ->
    // Sending messages
    tx.publisher().send("key1", "value1")
    tx.publisher().send("key2", "value2")

    // Commit consumer offsets within the transaction (exactly-once semantics)
    val offsets: Map<TopicPartition, OffsetAndMetadata> = ...
    val groupMetadata: ConsumerGroupMetadata = ...
    tx.sendOffsetsToTransaction(offsets, groupMetadata)

    // Explicit flush to guarantee sending before commit
    tx.flush()

    // commit() is called automatically on use close
}

Transaction<P> methods:

Method	Description
publisher()	Returns typed publisher for sending messages
producer()	Returns raw Producer<byte[], byte[]> for low-level operations
sendOffsetsToTransaction(offsets, groupMetadata)	Commits consumer offsets within the same transaction
flush()	Guarantees all messages are sent before commit
abort()	Aborts the transaction
abort(cause)	Aborts the transaction with specified cause
close()	Closes the transaction (commits if no abort)

A typical exactly-once pipeline reads a record, sends the result and the consumer offset inside one transaction, and therefore does not commit the offset through the Consumer at all:

Java Kotlin

@Component
public final class TransactionalPipelineListener {

    private final MyTransactionalPublisher publisher;

    public TransactionalPipelineListener(MyTransactionalPublisher publisher) {
        this.publisher = publisher;
    }

    @KafkaListener("kafka.someConsumer")
    public void process(ConsumerRecord<String, String> record, Consumer<String, String> consumer) {
        publisher.withTx(transaction -> {
            transaction.publisher().send("processed:" + record.value());
            transaction.sendOffsetsToTransaction(
                Map.of(new TopicPartition(record.topic(), record.partition()), new OffsetAndMetadata(record.offset() + 1)),
                consumer.groupMetadata()
            );
        });
    }
}

@Component
class TransactionalPipelineListener(private val publisher: MyTransactionalPublisher) {

    @KafkaListener("kafka.someConsumer")
    fun process(record: ConsumerRecord<String, String>, consumer: Consumer<String, String>) {
        publisher.begin().use { transaction ->
            transaction.publisher().send("processed:${record.value()}")
            transaction.sendOffsetsToTransaction(
                mapOf(TopicPartition(record.topic(), record.partition()) to OffsetAndMetadata(record.offset() + 1)),
                consumer.groupMetadata()
            )
        }
    }
}

Such a consumer should read with isolation.level = read_committed and enable.auto.commit = false, otherwise aborted transactions become visible or the offset is committed outside the transaction.

Transaction methods

TransactionalPublisher provides 4 methods for working with transactions:

Method	Passes to callback	Returns value
inTx(TransactionalConsumer)	P publisher	void
inTx(TransactionalFunction)	P publisher	R
withTx(TransactionConsumer)	Transaction<P> tx	void
withTx(TransactionFunction)	Transaction<P> tx	R

Any exception thrown from the callback aborts the transaction and is rethrown to the caller.

Example with return value:

Java Kotlin

// inTx with return value
Long messageId = transactionalPublisher.inTx(publisher -> {
    publisher.send("key", "value");
    return System.currentTimeMillis();
});

// withTx with Transaction access
transactionalPublisher.withTx(tx -> {
    tx.publisher().send("key", "value");
    tx.sendOffsetsToTransaction(offsets, groupMetadata);
    tx.flush(); // Explicit flush
});

// inTx with return value
val messageId = transactionalPublisher.inTx(TransactionalFunction<MyPublisher, RuntimeException, Long> { publisher ->
    publisher.send("key", "value")
    System.currentTimeMillis()
})

// withTx with Transaction access
transactionalPublisher.withTx(TransactionConsumer<MyPublisher, RuntimeException> { tx ->
    tx.publisher().send("key", "value")
    tx.sendOffsetsToTransaction(offsets, groupMetadata)
    tx.flush() // Explicit flush
})

Telemetry

Kafka uses a telemetry contract for logging, metrics, and tracing of messages. Telemetry configuration (the telemetry { logging / metrics / tracing } section) is described in the Configuration section.

KafkaPublisherTelemetryFactory builds a KafkaPublisherTelemetry for each publisher out of the publisher configuration path, the publisher interface name, KafkaPublisherTelemetryConfig, and driverProperties. KafkaPublisherTelemetry opens an observation per send and per transaction:

public interface KafkaPublisherTelemetry {

    MeterRegistry meterRegistry();

    KafkaPublisherTransactionObservation observeTx();

    KafkaPublisherRecordObservation observeSend(String topic);
}

KafkaPublisherRecordObservation also implements org.apache.kafka.clients.producer.Callback, so the driver completes it as soon as the broker acknowledges the record; the observation carries the topic, partition, offset, and send duration. KafkaPublisherTransactionObservation records the offsets sent into the transaction, commits, and rollbacks.

The default implementation is DefaultKafkaPublisherTelemetryFactory, registered by KafkaModule as a @DefaultComponent. It combines DefaultKafkaPublisherLoggerFactory for logging and DefaultKafkaPublisherMetricsFactory for metrics, both injected as optional dependencies so that either can be replaced by providing your own @Component subclass. Providing your own KafkaPublisherTelemetryFactory component replaces telemetry entirely.

Metrics and tracing are described in the Metrics Reference section.