Kora Kafka Producer

Kora Kafka Producer

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.

Kora generates the Kafka Producer implementation at compile time from an interface annotated with @KafkaPublisher. You declare the contract; the annotation processor (Java) or KSP (Kotlin) emits $Name_Impl, $Name_PublisherModule and $Name_TopicConfig, and wires the KafkaProducer, the Serializer<T>s and the telemetry.

Read this first when:

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

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

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

Not renamed, and still supported — do not "fix" these:

• @KafkaPublisher.Topic still exists and is still nested inside @KafkaPublisher.
• Future<RecordMetadata>, CompletionStage<RecordMetadata> and CompletableFuture<RecordMetadata> are still accepted publisher return types, and Kotlin suspend fun / Deferred<RecordMetadata> still work. The publisher is the one place Kora 2.0 did not drop asynchronous signatures — see Send signatures for the proof and the one extra dependency Kotlin needs.
• TransactionalPublisher<P> keeps its shape, inTx, begin() and the idPrefix / maxPoolSize / maxWaitTime config keys.

Quick Start

1. Dependencies

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

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

    implementation "io.koraframework:kafka"
    implementation "io.koraframework:json-common"   // only if you publish @Json payloads
}

Artifacts pulled through the koraBom platform inherit their version — never pin them individually. The Kotlin ksp configuration is not covered by the platform, so the processor there carries an explicit version.

2. Enable the module

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

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

JsonModule is io.koraframework.json.common.JsonModule; KafkaModule is io.koraframework.kafka.common.KafkaModule.

3. Declare a publisher

import io.koraframework.json.common.annotation.Json;
import io.koraframework.kafka.common.annotation.KafkaPublisher;
import io.koraframework.kafka.common.annotation.KafkaPublisher.Topic;

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

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

    @Topic(".topic")
    void send(@Json UserEvent event);
}

@KafkaPublisher takes the producer config path. @Topic takes a topic config path; a value starting with . is resolved relative to the publisher path, so .topic above means kafka.producer.userEvents.topic.

4. 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", Instant.now()));
    }
}

Constructor injection only — Kora has no field injection.

5. Configure

kafka {
  producer {
    userEvents {
      driverProperties {
        "bootstrap.servers": ${KAFKA_BOOTSTRAP}   # required
        "acks": "all"
      }
      topic {                                     # matches @Topic(".topic")
        topic = "user-events"                     # required
        # partition = 0                           # optional
      }
      telemetry.logging.enabled = true            # defaults to false
    }
  }
}

driverProperties is required and takes plain Kafka producer properties. Serializers for @Json and @Tag parameters come from the DI graph — do not set key.serializer / value.serializer in driverProperties.

Send signatures

K is the key type, V the value type. Parameters are classified by type: a org.apache.kafka.common.header.Headers parameter is headers, a org.apache.kafka.clients.producer.Callback parameter is a callback, a org.apache.kafka.clients.producer.ProducerRecord parameter is the whole record, and everything else is payload — one payload parameter is the value, two are key then value.

void is not fire-and-forget. Verified in the generated $X_Impl: every non-future signature ends in this.delegate.send(_record, _observation).get() wrapped in a try/catch that rethrows as KafkaPublishException. If you want fire-and-forget, take a Future and ignore it.

Parameter shapes, all valid with @Topic:

@Topic(".topic") void send(V value);
@Topic(".topic") void send(K key, V value);
@Topic(".topic") void send(K key, V value, Headers headers);
@Topic(".topic") void send(V value, Callback callback);

Without @Topic the method must take a single ProducerRecord<K, V> (the topic comes from the record), optionally followed by a Callback:

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

One shape per interface

Do not mix @Topic methods and ProducerRecord methods in the same @KafkaPublisher interface. Split them into two interfaces (they may share the same config path).

The generated $X_TopicConfig record and the $X_PublisherModule factory that fills it disagree on arity as soon as an interface mixes the two shapes. In Java this always fails:

error: constructor $MixedPublisher_TopicConfig in record $MixedPublisher_TopicConfig
       cannot be applied to given types;
  required: TopicConfig,TopicConfig
  found:    TopicConfig

In Kotlin it fails only when a non-@Topic method is declared before a @Topic one, which makes it look like a random ordering bug:

$KMixed2_PublisherModule.kt:37:5 Syntax error: Expecting an argument.

Kotlin suspend / Deferred publishers additionally need:

implementation("org.jetbrains.kotlinx:kotlinx-coroutines-jdk8:1.10.2")

Without it KSP generates fine and compileKotlin fails on the generated file with Unresolved reference 'kotlinx' / Unresolved reference 'await'.

See Producer Reference for the full API and config tree.

Serialization

For every key and value the processor resolves a org.apache.kafka.common.serialization.Serializer<T> from the graph:

1. @Json on the parameter → Kora's compile-time JSON serializer (needs io.koraframework:json-common and @Json on the DTO).
2. @Tag(SomeTag.class) on the parameter → the Serializer<T> @Component bound under that tag.
3. Neither → the untagged Serializer<T> for that type. KafkaSerializersModule supplies String, byte[], ByteBuffer, Bytes, Double, Float, Integer, Long, Short, UUID and Void.

Both annotations also work on a ProducerRecord type argument:

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

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

    @Topic(".topic")
    void send(String key, @Json MyEvent value);
}

@KafkaPublisher("kafka.producer.myPublisher")
public interface MyRecordPublisher {
    void send(ProducerRecord<String, @Json MyEvent> record);
}

A hand-written serializer over Kora's generated JsonWriter<T> must not wrap toByteArray in a try/catch (IOException) — in 2.0 the method declares no checked exception and the catch is a compile error (exception IOException is never thrown in body of corresponding try statement).

See Serialization Reference.

Transactional publishing

Declare the payload publisher, then a second @KafkaPublisher interface extending TransactionalPublisher<P> where P is that publisher. The migrated examples nest P inside the transactional interface so both fit in one file:

import io.koraframework.kafka.common.annotation.KafkaPublisher;
import io.koraframework.kafka.common.annotation.KafkaPublisher.Topic;
import io.koraframework.kafka.common.producer.TransactionalPublisher;

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

    @KafkaPublisher("kafka.producer.myPublisher")
    interface TopicPublisher {
        @Topic("kafka.producer.myTopic")
        void send(String value);
    }
}

inTx commits when the lambda returns and aborts when it throws:

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

kafka {
  producer {
    myTransactional {          # only these three keys are read here
      idPrefix = "order-service"   # transactional.id = "<idPrefix>-<random UUID>"
      maxPoolSize = 10
      maxWaitTime = "10s"
    }
    myPublisher {              # the wrapped publisher supplies the driver properties
      driverProperties { "bootstrap.servers": ${KAFKA_BOOTSTRAP} }
      myTopic { topic = "my-topic" }
    }
  }
}

The transactional section carries no driverProperties. Its config path is mapped to KafkaPublisherConfig.TransactionConfig, which declares only idPrefix, maxPoolSize and maxWaitTime; the transactional producers are built from the wrapped publisher's config plus a generated transactional.id. A driverProperties block under the transactional section is dead config that is silently ignored.

In Kotlin inTx is overloaded (consumer and function forms), so the SAM type must be named:

publisher.inTx(TransactionalConsumer<MyTransactionalPublisher.TopicPublisher, RuntimeException> { producer ->
    producer.send("""{"username":"Foo"}""")
})

A Kafka transaction covers Kafka only. It does not make a database write and a Kafka send atomic together — use the transactional outbox pattern. See Transactions Reference.

Telemetry

TelemetryConfig defaults in 2.0 are logging.enabled = false, metrics.enabled = false, tracing.enabled = true. A publisher emits no logs and no metrics until you turn them on per publisher section:

kafka.producer.myPublisher.telemetry {
  logging.enabled = true
  metrics.enabled = true
  metrics.driverMetrics = true   # binds Micrometer KafkaClientMetrics to the raw producer
}

KafkaModule supplies KafkaPublisherTelemetryFactory as a @DefaultComponent, so telemetry is customised by replacing components, not by registering listener factories the way 1.x did:

All three live in io.koraframework.kafka.common.producer.telemetry (…telemetry.impl for the Default* classes). There is no KafkaProducerLogger-style interface to implement any more.

References

Common pitfalls

Assets

Templates and config live in assets/.

scripts/generate_producer.py scaffolds a publisher, its HOCON section and an optional test; scripts/validate_config.py checks a kafka { … } section; scripts/run_evals.py prints the eval rubric.