Kora Kafka Consumer

Kora Kafka Consumer

Kora sub-skill — obey the kora-v2 meta rules on every task: R0 ground the workspace on Kora 2.0 refs before starting (framework source at tag 2.0.0.RC1 + kora-examples at migration/2.0; kora-docs is 1.x only) · R1 read this sub-skill before writing code · R2 Kora 2.0 APIs only — no Spring/Micronaut/Quarkus, no Kora 1.x APIs, no invented annotations or config keys · R3 journal any incorrect Kora usage. Add comments/Javadoc only if asked.

@KafkaListener turns a method on a @Component into a Kafka consumer. The annotation processor (Java) or KSP (Kotlin) generates a <ListenerClass>Module next to your class containing the config binding, the record handler and a @Root container — the graph starts and stops it for you.

Listener contracts are synchronous. Listener methods return void/Unit and run on a dedicated poll thread. There is no Kora Context — that type no longer exists anywhere in the framework.

An async return type on a listener is not rejected, it is silently discarded: neither consumer generator inspects the return type at all (no getReturnType / isFuture / isCompletionStage anywhere in KafkaConsumerHandlerGenerator.java or KafkaHandlerGenerator.kt), and the call is emitted as the bare statement controller.process(...). So a listener returning CompletionStage/Mono/Future compiles, and RecordHandler commits offset + 1 as soon as the method returns — before the async work finishes. The record is marked done while it is still in flight, and a failure inside the future never reaches telemetry or the backoff path. Return void/Unit and do the work inline.

This is the opposite of the publisher side, where an async return type is a first-class, generated contract — Future/CompletionStage (and, in Kotlin, suspend/Deferred) are the delivery acknowledgement of a send. Do not carry either rule across. See kora-kafka-producer.

suspend is the one exception worth knowing, and it is a trap rather than a feature: unlike the HTTP-server and repository processors, KSP does not reject a suspend listener — for the key/value and batch shapes it wraps your call in kotlinx.coroutines.runBlocking(Dispatchers.Unconfined) (KafkaHandlerGenerator.kt:133,236, with RC1 tests testProcessValueSuspend and testProcessRecordsSuspend). But kotlinx-coroutines appears in no Kora 2.0 build file, so your project must supply it; the coroutine is run to completion on the poll thread, so it buys no concurrency; and the single-ConsumerRecord shape has no such handling at all, so a suspend listener taking a ConsumerRecord generates Kotlin that does not compile. Write plain fun.

Renamed in 2.0 — check this before touching ported 1.x code

References: Consumer config · Listener signatures · Strategies · Serialization · Errors · Offsets · Batch · Rebalance · Telemetry · Transactions · Testing

Quick start

1. Dependencies — every Kora artifact takes its version from the BOM; never pin one individually.

configurations {
    koraBom
    annotationProcessor.extendsFrom(koraBom); implementation.extendsFrom(koraBom)
    testImplementation.extendsFrom(koraBom); testAnnotationProcessor.extendsFrom(koraBom)
}

dependencies {
    koraBom platform("io.koraframework:kora-bom:$koraVersion")   // koraVersion=2.0.0.RC1
    annotationProcessor "io.koraframework:annotation-processors"

    implementation "io.koraframework:kafka"
    implementation "io.koraframework:json-common"
    implementation "io.koraframework:config-hocon"
    implementation "io.koraframework:logging-logback"

    testImplementation "io.koraframework:test-junit5"
}

Kotlin uses ksp("io.koraframework:symbol-processors") instead of annotationProcessor — see kora-project-setup-kotlin.

2. Application graph — a @KoraApp interface extends each module (an interface never implements). The generated <ListenerClass>Module is discovered automatically inside the same compilation; do not add it to the extends clause.

@KoraApp
public interface Application extends HoconConfigModule, LogbackModule, JsonModule, KafkaModule {

    static void main(String[] args) {
        KoraApplication.run(ApplicationGraph::graph);
    }
}

3. Listener

@Component
public final class UserEventListener {

    private static final Logger log = LoggerFactory.getLogger(UserEventListener.class);

    @KafkaListener("kafka.consumer.user-events")
    void process(String value) {
        log.info("Received: {}", value);
    }
}

4. Configuration — the @KafkaListener value is the full config path, nothing is prefixed for you. kafka.consumer.<name> is a convention used by the Kora examples, not a framework rule.

kafka {
  consumer {
    user-events {
      topics = ["user-events"]
      driverProperties {
        "bootstrap.servers" = "localhost:9092"
        "bootstrap.servers" = ${?KAFKA_BOOTSTRAP}
        "group.id" = "user-service"
        "auto.offset.reset" = "earliest"
      }
    }
  }
}

HOCON has no ${VAR:default} placeholder. Write the literal first, then override it with ${?VAR}.

Accepted handler signatures

The processor classifies each parameter by type, then picks one of three handler shapes. Anything it cannot classify becomes a payload parameter — and a third payload parameter is a compile error (Kafka listener method has too many payload parameters). Inject services through the class constructor, never as a listener parameter.

Batch listeners accept only ConsumerRecords and Consumer — a Headers or exception parameter there fails with Kafka records listener method has unsupported parameter.

RC1, Kotlin only: a listener that takes both Headers and a deserialization-exception parameter generates code that does not compile — KSP emits val headers = record.headers() inside the generated try block and then reads it outside. Java is unaffected, and Kotlin Headers without an exception parameter is fine. Fixed after RC1 on master (8e1eec9a4). Until a release carries it, take ConsumerRecord and call record.headers() yourself.

void process(String value)                                             // value only
void process(String key, String value)                                 // key, value
void process(String key, String value, Headers headers)                // + headers
void process(ConsumerRecord<String, String> record)                    // full metadata
void process(ConsumerRecords<String, String> records)                  // whole poll batch
void process(ConsumerRecord<String, String> r, Consumer<String, String> c)  // manual commit
void process(@Nullable @Json Event e, @Nullable Exception exception)   // deserialization errors

The complete list, with the processor test that proves each one, is in the listener reference.

Who commits the offset

Kora decides this at compile time from the signature, and at runtime from enable.auto.commit.

If enable.auto.commit is absent the subscribe container forces it to false and takes over committing. Details and patterns: offsets.

Subscribe vs assign

The container is chosen by the presence of group.id in driverProperties:

At RC1 the generated assign container rejects anything but a single topic: @KafkaListener require to specify 1 topic to subscribe when groupId is null, but received: ... (Kotlin fails a bare require(topics.size == 1)). A topicsPattern without a group.id leaves topics null and hits the same check. Multi-topic assign landed after RC1 on master (a0a2b8c1d) — do not write it against a released version.

Strategies reference.

Deserialization

KafkaDeserializersModule supplies Deserializer beans for String, UUID, byte[], Bytes, ByteBuffer, Double, Float, Integer, Long, Short and Void. @Json on the payload type selects the generated JsonKafkaDeserializer<T>; @Tag(X.class) selects your own Deserializer<T> @Component carrying that tag.

@Json public record OrderEvent(String orderId, BigDecimal amount) {}

@KafkaListener("kafka.consumer.orders")
void process(@Nullable @Json OrderEvent event, @Nullable Exception exception) {
    if (exception != null) { log.warn("bad record", exception); return; }
    orderService.process(event);
}

Kotlin writes the nullability into the type, not into an annotation, and JsonReader.read is @Nullable:

@KafkaListener("kafka.consumer.orders")
fun process(@Json event: OrderEvent?, exception: Exception?) { ... }

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

Serialization reference.

Errors

Deserialization is lazy: ConsumerRecord.key()/value() throw RecordKeyDeserializationException / RecordValueDeserializationException, and both carry getRecord() returning the raw ConsumerRecord<byte[], byte[]> for a DLQ. Error handling reference.

Configuration keys

kafka.consumer.<name> is the conventional path. Everything below is KafkaListenerConfig.

Logging and metrics are off by default in 2.0. An example that claims to demonstrate them must turn them on explicitly. Consumer config reference · Telemetry reference.

Common pitfalls

Templates and scripts

assets/ — Application.{java,kt}.template, ConsumerListener.{java,kt}.template, JsonMessageListener.{java,kt}.template, ConsumerListenerTests.{java,kt}.template, MessagePublisher.{java,kt}.template, JsonMessagePublisher.{java,kt}.template, TransactionalPublisher.{java,kt}.template, MessagePublisherTests.{java,kt}.template, application.conf.template, build.gradle.template, build.gradle.kts.template.

scripts/ — generate_consumer.py, generate_producer.py (both support --dry-run) and validate_config.py (read-only). Run --help for the full CLI.

python3 scripts/generate_consumer.py --name UserEvent --topics user-events \
    --package com.example --signature record --dry-run
python3 scripts/validate_config.py --config src/main/resources/application.conf

Related skills

• kora-kafka-producer — @KafkaPublisher, @Topic, transactions
• kora-json — @Json, JsonReader/JsonWriter
• kora-di-compile — @KoraApp, @Component, @Tag
• kora-config-hocon — HOCON substitution rules
• kora-testing-junit-java · kora-testing-junit-kotlin — @KoraAppTest
• kora-telemetry-metrics — the meters this skill configures
• kora-aop-resilient — @Retryable on a listener method

Source of truth

Kora 2.0 has no documentation site. Any page served under a /v2/ path is a copy of the 1.x content and describes ru.tinkoff.kora. Resolve 2.0 questions from:

• Framework source, tag 2.0.0.RC1: kafka · kafka-annotation-processor · kafka-symbol-processor
• Migrated examples, branch migration/2.0: kora-java-kafka · kora-kotlin-kafka
• Migrated guide apps, branch migration/2.0: kora-java-guide-messaging-kafka-app · kora-kotlin-guide-messaging-kafka-app
• Apache Kafka consumer configs for everything inside driverProperties