Kora Kafka Consumer

Kora Kafka Consumer Skill

Kora sub-skill — obey the kora-v1 meta rules on every task: R0 ensure .kora-agent/ docs+examples are cloned · R1 read this sub-skill before writing code · R2 Kora APIs only — no Spring/Micronaut/Quarkus, no invented annotations or config keys · R3 journal any incorrect Kora usage. Add comments/Javadoc only if asked.

Languages: Java, Kotlin | Build: Gradle

Level 1 — Quick Start | Level 2 — Signatures | Level 3 — Errors & Offset | Level 4 — Batch & Rebalance

References: Consumer config | Listener signatures | Strategies | Serialization | Errors | Offset | Batch | Rebalance | Telemetry | Transactions (producer) | Testing

Quick Start

1. Dependencies (Kora artifacts inherit their version from the kora-parent BOM — never pin them individually):

dependencies {
    koraBom platform("ru.tinkoff.kora:kora-parent:1.2.19")
    annotationProcessor "ru.tinkoff.kora:annotation-processors"

    implementation "ru.tinkoff.kora:kafka"
    implementation "ru.tinkoff.kora:json-module"
    implementation "ru.tinkoff.kora:config-hocon"
    implementation "ru.tinkoff.kora:logging-logback"
}

2. Application Module (a @KoraApp interface extends each module — interfaces never use implements):

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

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

3. Simple Listener:

@Component
public final class UserEventListener {
    @KafkaListener("kafka.consumer.userEvents")
    void process(String value) {
        log.info("Received: {}", value);
    }
}

4. Configuration:

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

Method Signatures

JSON with Error Handling:

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

@KafkaListener("kafka.consumer.orders")
void process(@Nullable @Json OrderEvent event, @Nullable Exception error) {
    if (error != null) {
        log.error("Deserialization failed", error);
        return;
    }
    orderService.process(event);
}

Full signatures: Listener Reference

Strategies

Subscribe (load balancing):

driverProperties { "group.id" = "order-service"; "bootstrap.servers" = "localhost:9092" }

Assign (broadcast):

driverProperties { "bootstrap.servers" = "localhost:9092" }  # No group.id

Details: Strategies Reference

Error Handling

Deserialization:

@KafkaListener("kafka.consumer.events")
void process(@Nullable @Json Event event, @Nullable Exception error) {
    if (error != null) { log.error("Failed", error); return; }
    processEvent(event);
}

Skip Invalid:

@KafkaListener("kafka.consumer.events")
void process(Event event) {
    if (event.orderId() == null)
        throw new KafkaSkipRecordException(new IllegalArgumentException("Missing orderId"));
    processEvent(event);
}

DLQ:

@KafkaListener("kafka.consumer.events")
void process(@Nullable Event event, @Nullable Exception error) {
    if (error != null) { dlqPublisher.send("dlq", event, error.getMessage()); return; }
    processEvent(event);
}

Details: Error Handling Reference

Offset Management

Auto (Default): Commit after each message/batch.

Manual:

@KafkaListener("kafka.consumer.events")
void process(ConsumerRecord<String, String> record, Consumer<String, String> consumer) {
    try { process(record.value()); consumer.commitSync(); }
    catch (Exception e) { throw e; }
}

Rebalance: provide a ConsumerAwareRebalanceListener as a @Component carrying the consumer's tag. Kora generates a tag per listener (<Listener>Module.<Listener>ProcessTag), or you can declare your own via @KafkaListener(value = "...", tag = MyTag.class) and reuse it:

@Tag(MyTag.class) @Component
final class RebalanceListener implements ConsumerAwareRebalanceListener {
    public void onPartitionsRevoked(Consumer<?, ?> c, Collection<TopicPartition> p) {
        c.commitSync();  // Commit before rebalance
    }
    public void onPartitionsAssigned(Consumer<?, ?> c, Collection<TopicPartition> p) { }
}

Details: Offset Reference

Batch Processing

@KafkaListener("kafka.consumer.orders")
void process(ConsumerRecords<String, OrderEvent> records) {
    for (ConsumerRecord<String, OrderEvent> record : records)
        orderService.process(record.value());
}

Config:

kafka.consumer.batchProcessor {
  topics = ["high-volume"]
  threads = 4
  driverProperties { "max.poll.records" = 500; "fetch.min.bytes" = 1048576 }
}

Details: Batch Reference

Rebalance Handling

@Tag(MyTag.class) @Component
final class MyRebalanceListener implements ConsumerAwareRebalanceListener {
    public void onPartitionsRevoked(Consumer<?, ?> c, Collection<TopicPartition> p) {
        log.info("Revoked: {}", p); c.commitSync(); cache.clear();
    }
    public void onPartitionsAssigned(Consumer<?, ?> c, Collection<TopicPartition> p) {
        log.info("Assigned: {}", p);
    }
    public void onPartitionsLost(Consumer<?, ?> c, Collection<TopicPartition> p) {
        log.warn("Lost: {}", p);  // Don't commit
    }
}

Details: Rebalance Reference

Configuration

Required:

kafka.consumer.myListener {
  topics = ["topic1"]
  driverProperties { "bootstrap.servers" = "localhost:9092" }
}

Optional:

Telemetry:

telemetry { logging {enabled=true}; metrics {enabled=true}; tracing {enabled=true} }

Common Pitfalls

Do not use field injection — Kora wires components through constructor injection at compile time, and every consumer is a @Component with @KafkaListener methods.

Templates

Assets: ConsumerListener.java.template, JsonMessageListener.java.template, ConsumerListenerTests.java.template, application.conf.template

See assets/ for generator script.

Testing

Details: Testing Reference

Source of truth: Kafka doc | Messaging guide | Example app