Kora Kafka Producer

Kora Kafka Producer

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.

Kora generates a Kafka Producer implementation at compile time from an interface annotated with @KafkaPublisher. You declare the contract; the annotation processor wires the KafkaProducer, serializers, and telemetry.

Read this first when:

• declaring a typed publisher with @KafkaPublisher,
• choosing a send signature (sync, RecordMetadata, async, ProducerRecord),
• selecting serializers with @Json or @Tag,
• sending atomically with TransactionalPublisher.

For Kafka consumers (@KafkaListener), use the kora-kafka-consumer skill.

Quick Start

1. Dependencies

dependencies {
    koraBom platform("ru.tinkoff.kora:kora-parent:1.2.19")
    annotationProcessor "ru.tinkoff.kora:annotation-processors" // KSP for Kotlin: ksp "ru.tinkoff.kora:symbol-processors"

    implementation "ru.tinkoff.kora:kafka"
    implementation "ru.tinkoff.kora:json-module" // only if you publish @Json payloads
}

All ru.tinkoff.kora:* artifacts inherit their version from the kora-parent BOM — never pin them individually.

2. Enable the module

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

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

3. Declare a publisher

import ru.tinkoff.kora.kafka.common.annotation.KafkaPublisher;
import ru.tinkoff.kora.kafka.common.annotation.KafkaPublisher.Topic;

@KafkaPublisher("kafka.producer.userEvents")
public interface UserEventPublisher {

    @Topic("kafka.producer.userEventsTopic")
    void send(@Json UserEvent event);
}

@KafkaPublisher points to the producer config section; @Topic (the nested @KafkaPublisher.Topic) points to a topic config section.

4. Define the event

@Json
public record UserEvent(String userId, String eventType, LocalDateTime timestamp) {}

5. Inject and publish

@Component
public final class UserService {

    private final UserEventPublisher publisher;

    public UserService(UserEventPublisher publisher) {
        this.publisher = publisher;
    }

    public void createUser(String userId) {
        publisher.send(new UserEvent(userId, "CREATED", LocalDateTime.now()));
    }
}

Use constructor injection — Kora has no field injection.

6. Configure

kafka {
  producer {
    userEvents {
      driverProperties {
        "bootstrap.servers": ${KAFKA_BOOTSTRAP}  # required
        "acks": "all"
      }
      telemetry.logging.enabled = true
    }
    userEventsTopic {
      topic = "user-events"  # required
      # partition = 0        # optional
    }
  }
}

The topic config section is a sibling of the producer section under kafka.producer, named to match the @Topic path. Serializers for @Json parameters are injected by Kora — do not set key.serializer / value.serializer in driverProperties for those parameters.

Send signatures

K is the key type, V the value type. With @Topic, value is required; key and Headers are optional. Available return types:

In Kotlin a suspend fun send(value: V): RecordMetadata and Deferred<RecordMetadata> are also supported.

A method WITHOUT @Topic must take a ProducerRecord (the topic comes from the record). A method WITH @Topic takes unpacked value/key/headers.

See Producer Reference for full detail.

Serialization

Kora chooses a Serializer<T> for each key/value from the application graph:

• @Json on a parameter (and on its DTO) serializes that value as JSON via the json-module.
• @Tag(SomeTag.class) on a parameter selects a custom Serializer<T> component bound under that tag.
• Otherwise the matching Serializer<T> component is used (Kora provides the standard String/byte/numeric serializers).

@KafkaPublisher("kafka.producer.myPublisher")
public interface MyPublisher {

    @Topic("kafka.producer.myTopic")
    void send(String key, @Json MyEvent value);

    // custom serializer selected by tag
    void send(ProducerRecord<String, @Tag(MyEvent.class) MyEvent> record);
}

See Serialization Reference.

Transactional publishing

To send several records atomically (all or nothing), declare a regular publisher, then a second @KafkaPublisher interface that extends TransactionalPublisher<P>:

import ru.tinkoff.kora.kafka.common.producer.TransactionalPublisher;

@KafkaPublisher("kafka.producer.myPublisher")
public interface MyPublisher {
    @Topic("kafka.producer.myTopic")
    void send(String key, String value);
}

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

Send inside inTx — on success the transaction commits, on any thrown exception it aborts:

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

Transaction config lives under the transactional publisher section:

kafka {
  producer {
    myTransactional {
      idPrefix = "order-service-"  # transactional.id prefix
      maxPoolSize = 10             # transaction connection pool size
      maxWaitTime = "10s"          # max transaction wait time
    }
  }
}

Kafka transactions cover Kafka only. They do NOT make a database write and a Kafka send atomic together — for that use the transactional outbox pattern. See Transactions Reference.

References

Common pitfalls

Assets

Templates and config live in assets/.