Skip to content
Docs

Kafka

View Markdown

Vertique’s Kafka integration connects topics to the typed execution model the rest of the framework uses. Consumed records dispatch into service contracts, and typed producer interfaces publish records.

  • Topics dispatch into service contracts. Annotate a service implementation method with @KafkaSource, or declare a consumer binding. Consumed records then go through the same service dispatch, resilience, and context propagation as every other call.
  • Typed producers. Annotate an interface with @KafkaProducer. Each method sends to its declared topic through a generated proxy, so the application writes no client code.
  • Four consumption models. They range from one annotation that binds a topic to an operation up to a fully custom record handler. Pick the lowest-coupling model that expresses the routing. Config overrides always win.
  • Explicit delivery semantics. The commit strategy decides what can be lost. AUTO is at-most-once. MANUAL is at-least-once and is required for retry and dead-letter handling, so handlers under MANUAL must be idempotent.
  • Dead-letter with provenance. A dead-lettered record keeps its original key, value, and headers, plus headers that record the source topic, partition, offset, consumer name, and error.
  • Format-neutral core. JSON and Avro serdes ship as separate modules. A consumer whose format has no registered provider fails at startup, before it reads a record.
  • Headers as they are on the wire. Handlers, filters, and producers exchange record headers as one immutable value that keeps order, repeated keys, and binary values.
  • One completion callback per record. An interceptor sees every record the consumer received together with its final outcome, including records that were filtered or never deserialized.

Correlation travels with the record. An x-correlation-id header seeds the dispatch’s correlation id, and the framework generates one when the header is absent.

Reach for this module when Kafka records should enter the typed service model. For a raw client with no dispatch, use vertx-kafka-client directly. Producing works together with inbox/outbox messaging: publish directly through a typed producer when the send does not need to commit with a business write, and record the send through the transactional outbox’s Kafka adapter when it does. The outbox relay publishes through this same module.

Consume a topic into a service operation

Section titled “Consume a topic into a service operation”

The lowest-coupling model is one annotation on a service implementation method. From the vertique-kafka-core module reference:

public class OrderServiceImpl implements OrderService {


    @KafkaSource(topic = "order.created", groupId = "order-processing")
    @Override
    public Future<Void> processOrder(OrderCreatedEvent event) {
        return Future.succeededFuture();
    }
}

The framework subscribes, deserializes the value with the registered format provider, and dispatches to the operation over the event bus. For full control (a custom deserializer, a pre-deserialization filter, an explicit dead-letter topic), declare a KafkaConsumerBinding through a Dagger multibinding instead. A routing interface and a custom record handler are the other two models. The module reference documents all four.

Produce through a typed interface

Section titled “Produce through a typed interface”

From the same module reference. Every method names its topic, and the second method takes the record key before the value:

@KafkaProducer
public interface OrderEvents {


    @Topic("order.created")
    Future<RecordMetadata> orderCreated(OrderCreatedEvent event);


    @Topic("order.created")
    Future<RecordMetadata> orderCreatedWithKey(String key, OrderCreatedEvent event);
}

KafkaProducerFactory.create(...) builds the proxy. The same factory has raw byte-level sends for cases a typed method cannot express.

Give every method on a producer interface its own name. The topic and serializer of a method are configured by that name, so create(...) throws IllegalArgumentException when two methods share one.

Read and send record headers

Section titled “Read and send record headers”

KafkaRecordHeaders holds the headers of one record in wire order, with repeated keys, null values, and binary values intact. The consumer extracts it once per record, and the filter, the deserializer, the interceptors, and the handler all read that same immutable instance.

A service operation reads headers from KafkaRecordContext, either as a handler parameter or through KafkaRecordContext.current(). A custom record handler reads them from its KafkaMessage. From the vertique-kafka-core module reference:

// One header, as on the wire. Empty only when the key is absent.
Optional<KafkaRecordHeader> signature = message.headers().lastHeader("signature");


// One text value per key. Call asMap() once and keep the map when you read several keys.
Map<String, String> text = message.headers().asMap();
String eventType = text.get("event-type");            // null when absent or null-valued

asMap() is lossy. A repeated key keeps its last non-null value, a header with a null value is left out, and malformed UTF-8 bytes are replaced. Use lastHeader or headers(key) when the exact bytes matter.

To send headers, declare KafkaRecordHeaders as the last parameter of a producer method. The module reference’s OrderEvents interface has two more methods for that:

@Topic("order.created")
Future<RecordMetadata> orderCreatedWithHeaders(OrderCreatedEvent event, KafkaRecordHeaders headers);


@Topic("order.created")
Future<RecordMetadata> orderCreatedWithKeyAndHeaders(
        String key, OrderCreatedEvent event, KafkaRecordHeaders headers);
// Text headers: one UTF-8 header per entry, in the map's iteration order.
orderEvents.orderCreatedWithHeaders(event, KafkaRecordHeaders.of(Map.of("event-type", "order.created")));


// Repeated keys and binary values, in the order given.
KafkaRecordHeaders headers = new KafkaRecordHeaders(List.of(
        KafkaRecordHeader.ofUtf8("x-forwarded-by", "gateway"),
        KafkaRecordHeader.ofUtf8("x-forwarded-by", "orders"),
        new KafkaRecordHeader("signature", Buffer.buffer(signatureBytes))));
orderEvents.orderCreatedWithKeyAndHeaders(orderId, event, headers);

The record carries your headers first, in the order you gave them. The framework then appends one vertique- prefixed header per durable-context namespace. Two kinds of header fail the returned future with IllegalArgumentException, and nothing is sent: a key that starts with the reserved vertique- prefix, and a header with a null value. A Map parameter is not accepted as headers. Build the value with KafkaRecordHeaders.of(map).

Observe every record’s outcome

Section titled “Observe every record’s outcome”

KafkaConsumerInterceptor.onRecordCompleted runs once for each delivered record, after its disposition is final and all asynchronous work for it has settled. It covers the paths the other interceptor callbacks miss: a record filtered before deserialization, one that matched no route, and one that failed to deserialize. Use it for audit trails, outcome counters, and alerts on lost records. From the module reference:

@Override
public void onRecordCompleted(KafkaConsumerCompletedEvent event, KafkaConsumerRecordView record) {
    KafkaConsumerRecordIdentity id = event.identity();
    completions.merge(event.outcome(), 1L, Long::sum);   // a ConcurrentHashMap: thread-safe
    if (event.outcome() == KafkaTerminalOutcome.DLQ_FAILED) {
        log.warn("Lost record {}-{}@{} key={}",
                id.topic(), id.partition(), id.offset(), record.key());
    }
}

The event carries the record’s coordinates, its retry count, and a KafkaTerminalOutcome such as SUCCESS, SKIP, RETRY_SCHEDULED, DLQ_PUBLISHED, or DLQ_FAILED. It holds no key, headers, or value, so it is safe to log as is. The view gives the key, the headers, and the raw value bytes.

Three constraints apply to an implementation:

  • The callback is an observer. The decision is already made, and it cannot change filtering, commit, retry, or dead-lettering.
  • It may run on any thread, and concurrently for different records. Keep the implementation thread-safe.
  • The value is the broker’s array, uncopied. Copy the bytes before keeping them past the callback.

A record redelivered for a retry completes once per delivery. Two cases produce no completion: an asynchronous interceptor or handler future that never completes, and a throw from inside an asynchronous continuation.

Contribute the interceptor through the same Dagger set multibinding as any other KafkaConsumerInterceptor. An exception thrown by the callback is logged and swallowed, and the record’s outcome is unaffected.

Delivery semantics

Section titled “Delivery semantics”
CommitStrategy Semantics Notes
AUTO (default) At-most-once Offsets auto-commit on an interval; a crash between commit and processing loses the record. Cannot be combined with retry or dead-letter handling
MANUAL At-least-once Offsets commit only after successful dispatch; records may be reprocessed but are never lost. Required by the retry and dead-letter error strategies, so handlers must be idempotent

An inbound record that must produce an exactly-once effect pairs MANUAL commit with inbox deduplication on the handler side.

Wiring

Section titled “Wiring”

Install KafkaModule plus at least one format module in your component: KafkaJsonModule (vertique-kafka-json) for JSON, or AvroModule (vertique-kafka-avro) for Avro. Consumers are not deployed automatically. The application calls the consumer deployment manager’s deployAll(), which registers codecs and deploys every enabled consumer verticle. If any deployment fails, it rolls back the whole batch.

Applications parented on vertique-app-parent get the annotation-processor facade automatically. The generated Kafka consumers module contributes @KafkaSource classes and annotated handler classes. Routing interfaces stay a manual contribution. The module reference’s auto-wiring section lists which is which.

Learn more

Section titled “Learn more”

Continue reading

Section titled “Continue reading”