Kafka core

View Markdown

Status: Stable Package: dev.vertique.kafka Artifact: vertique-kafka-core Depends on: core, resilience, context, logging, deploy, services API reference: javadoc.io

vertique-kafka-core bridges Kafka topics to Vertique services. Declare a consumer with an annotation or a builder and the framework subscribes to the topic, deserializes the value, and dispatches it over the event bus to a service contract operation with framework context propagation intact. Producing works the same way in reverse: annotate an interface and call typed methods.

The module is format-neutral. It ships no serializer or deserializer of its own — add dev.vertique:vertique-kafka-json for JSON, dev.vertique:vertique-kafka-avro for Avro, or contribute your own KafkaSerdeProvider. A consumer whose resolved format has no registered provider fails at startup rather than at the first record.

When To Use It

Install vertique-kafka-core when an application consumes from or produces to Kafka and wants records routed into the same service-dispatch, resilience, and context-propagation model the rest of the framework uses. If you only need a raw Kafka client, use io.vertx:vertx-kafka-client directly.

Always pair it with at least one format module:

Need	Artifacts	`@Component` modules
Raw handlers or a custom format only	`vertique-kafka-core`	`KafkaModule` plus your provider module
JSON	`vertique-kafka-core`, `vertique-kafka-json`	`KafkaModule`, `KafkaJsonModule`
JSON and Avro	`vertique-kafka-core`, `vertique-kafka-json`, `vertique-kafka-avro`	`KafkaModule`, `KafkaJsonModule`, `AvroModule`

Core Concepts

Four consumption models

Pick the lowest-coupling model that expresses the routing you need.

Model	Declared with	Use when
1 — Service source	`@KafkaSource` on a service impl method	One topic feeds one existing service operation
2 — Declarative binding	`KafkaConsumerBinding` multibinding	You need full config control, a custom deserializer, or a pre-deserialization filter
3 — Routing interface	`@KafkaListener` interface with `@KafkaHandler` methods	One topic routes to several service operations by header or JSON property
4 — Custom handler	`@KafkaListener` on a `KafkaRecordHandler` class	You need arbitrary dispatch, transformation, or multi-service fan-out

Whatever the model, config at kafka.consumers.{name}.* always wins over annotation and builder values.

Delivery semantics

Delivery is decided by the pair of CommitStrategy and ErrorStrategy. The commit strategy is the one that determines whether a record can be lost.

`CommitStrategy`	Semantics	Notes
`AUTO` (default)	At-most-once	Kafka auto-commits on an interval. A crash between commit and processing loses the record. Cannot be combined with `DEAD_LETTER` or `RETRY`
`MANUAL`	At-least-once	Offsets commit only after a successful dispatch. Records may be reprocessed but are never lost. Required by `DEAD_LETTER` and `RETRY`

Handlers must therefore be idempotent under MANUAL.

Record headers the framework reads and writes

Header	Direction	Meaning
`x-correlation-id`	read on consume	Seeds the record’s correlation id. A record without it gets a generated UUID. Bound to MDC as `kafka.correlationId` for the dispatch
the `matchHeader` of a route	read on consume	Selects a model-3 route
`vertique-<namespace>`	read on consume	Decoded into the durable context that is bound for the dispatch. A value that is not a JSON object is skipped
`x-dlq-source-topic`	written on DLQ publish	Original topic
`x-dlq-source-partition`	written on DLQ publish	Original partition
`x-dlq-source-offset`	written on DLQ publish	Original offset
`x-dlq-consumer`	written on DLQ publish	Name of the consumer that failed
`x-dlq-error`	written on DLQ publish	Exception simple name, plus up to 200 characters of its message
`vertique-<namespace>`	written on every send except the DLQ publish	One text header per durable-context namespace, value = the namespace body as JSON. Appended after the application headers. The `vertique-` prefix is reserved: an application header that uses it fails the send

The framework reads each of those headers as text: the last value of the key that is not null, decoded as UTF-8 with malformed bytes replaced. The KafkaRecordFilter factories read headers the same way. That is the rule of KafkaRecordHeaders.asMap().

A dead-lettered record keeps its original key and its original value bytes, and its headers are the failed record’s headers as received — in order, with repeated keys and binary values byte for byte — followed by the five x-dlq-* headers. Their names are constants on KafkaDlqHeaders. Two kinds of original header are left out:

one whose key is exactly one of the five x-dlq-* names, so a record that is dead-lettered a second time carries only the new ones;
one with a null value, because a producer record cannot carry it.

The original headers include the framework’s own vertique-<namespace> context headers. They are forwarded as they were received, so the dead-letter record carries the failed record’s context. No context is captured from the scope that handles the error, and no context header that the failed record did not carry is added.

Getting Started

@Singleton
@Component(modules = {
    VertxModule.class, ConfigModule.class, DispatchModule.class,
    KafkaModule.class,      // core runtime
    KafkaJsonModule.class,  // JSON format provider, from vertique-kafka-json
    AppModule.class, ServiceModule.class
})
interface AppComponent {
    KafkaConsumerDeploymentManager kafkaConsumerDeploymentManager();
    KafkaProducerFactory kafkaProducerFactory();
}

Consumers are not deployed automatically — the application starts them:

component.kafkaConsumerDeploymentManager().deployAll();

deployAll() registers local codecs, then deploys every enabled consumer verticle in parallel; if any deployment fails the successfully-started consumers are rolled back and the returned future fails with the original cause. undeployAll() is the counterpart. Consumer verticles are deployed in the SERVICES lifecycle phase.

Model 1 — `@KafkaSource`

Place it on a service implementation method. The registrar builds the binding and dispatches records to that operation.

public class OrderServiceImpl implements OrderService {

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

Attribute	Default	Description
`name`	`""` — derived, see below	Config key used for overrides
`topic`	required	Topic to consume
`groupId`	required	Consumer group id
`errorStrategy`	`SKIP`	Failure handling
`commitStrategy`	`AUTO`	Offset commit timing
`deadLetterTopic`	`""` → `"{topic}.dlq"`	DLQ topic for `DEAD_LETTER`

A derived name is "{serviceName}-{operationId}", or "{namespace}-{serviceName}-{operationId}" when the service declares a namespace, so same-named services in different namespaces do not collide.

Model 2 — `KafkaConsumerBinding`

@Provides @IntoSet
static KafkaConsumerBinding<?> orderBinding() {
    return KafkaConsumerBinding.builder("order-created", OrderCreatedEvent.class)
            .topic("order.created")
            .groupId("order-processing")
            .dispatchTo(OrderService.class, "processOrder")
            .errorStrategy(ErrorStrategy.DEAD_LETTER)
            .commitStrategy(CommitStrategy.MANUAL)
            .filter(KafkaRecordFilter.headerEquals("event-type", "order.created"))
            .build();
}

Builder method	Default	Description
`builder(String name, Class<V> valueType)`	required	Binding name and deserialization target
`topic(String)`	required	Topic
`groupId(String)`	required	Consumer group id
`dispatchTo(Class<?> service, String operation)`	required	Target service contract and operation
`errorStrategy(ErrorStrategy)`	`SKIP`	Failure handling
`commitStrategy(CommitStrategy)`	`AUTO`	Offset commit timing
`deadLetterTopic(String)`	`"{topic}.dlq"`	DLQ topic
`deserializer(KafkaDeserializer<V>)`	framework-built from the resolved format	Custom deserializer; bypasses format and profile resolution entirely
`filter(KafkaRecordFilter)`	`null` — accept all	Pre-deserialization filter on key and headers
`enabled(boolean)`	`true`	Deploy at startup
`eventBusTimeoutMs(long)`	`30000`	Event-bus dispatch timeout
`jsonProfile(JsonProfileId)`	`null` — framework default	JSON mapper profile for value deserialization

Use the durable operation id (@ServiceOperation value) in dispatchTo for stable-target-eligible operations so the binding survives a method rename.

Model 3 — `@KafkaListener` routing interface

@KafkaListener(name = "order-events", topic = "order.events", groupId = "order-processing")
public interface OrderEventRouter {

    @KafkaHandler(matchHeader = "event-type", matchValue = "order.created")
    @DispatchTo(service = OrderService.class, operation = "processOrder")
    void onOrderCreated(OrderCreatedEvent event);

    @KafkaHandler(matchHeader = "event-type", matchValue = "order.cancelled")
    @DispatchTo(service = OrderService.class, operation = "cancelOrder")
    void onOrderCancelled(OrderCancelledEvent event);

    @KafkaHandler(defaultHandler = true)
    void onUnmatched();
}

A routing interface contributes its Class<?> literal, not an instance:

@Provides @IntoSet @KafkaConsumers
static Object orderEventRouter() {
    return OrderEventRouter.class;
}

`@KafkaListener` attribute	Default	Description
`name`	required	Binding name used for config overrides
`topic`	required	Topic
`groupId`	required	Consumer group id
`valueType`	`Void.class`	Deserialization target for model 4; inferred per method in model 3
`errorStrategy`	`SKIP`	Failure handling
`commitStrategy`	`AUTO`	Offset commit timing
`deadLetterTopic`	`""` → `"{topic}.dlq"`	DLQ topic

`@KafkaHandler` attribute	Default	Description
`matchHeader`	`""`	Header name to match; mutually exclusive with `matchProperty` and `defaultHandler`
`matchProperty`	`""`	JSON property name to match after deserialization
`matchValue`	`""`	Value the header or property must equal
`defaultHandler`	`false`	Catches unmatched records; at most one per listener

`@DispatchTo` attribute	Description
`service`	Target service contract interface
`operation`	Durable operation id

Omitting @DispatchTo consumes the matched record without dispatching it.

Model 4 — `@KafkaListener` on a `KafkaRecordHandler`

@KafkaListener(name = "order-events", topic = "order.events",
               groupId = "order-processing", valueType = OrderEvent.class)
public class OrderEventHandler implements KafkaRecordHandler<OrderEvent> {

    private final OrderService orderService;
    private final InventoryService inventoryService;

    @Inject
    public OrderEventHandler(OrderService orderService, InventoryService inventoryService) {
        this.orderService = orderService;
        this.inventoryService = inventoryService;
    }

    @Override
    public Future<Void> handle(KafkaMessage<OrderEvent> message) {
        return switch (message.headers().asMap().getOrDefault("event-type", "")) {
            case "order.created" -> orderService.processOrder(message.value());
            case "inventory.reserved" -> inventoryService.confirmReservation(message.value());
            default -> Future.succeededFuture();
        };
    }
}

@Provides @IntoSet @KafkaConsumers
static Object orderEventHandler(OrderEventHandler handler) {
    return handler;
}

Generated auto-wiring

Applications inheriting vertique-app-parent declare the Kafka runtime capability they use and get the processor facade automatically; custom-parent applications follow the BOM plus vertique-codegen-all recipe in docs/packaging.md. dev.vertique:vertique-codegen-dagger emits @Provides @IntoSet @KafkaConsumers Object bindings for @KafkaSource-annotated classes (model 1) and @KafkaListener-annotated classes (model 4); include GeneratedKafkaConsumersModule in the @Component.

Model 3 routing interfaces are not auto-wired — their Class<?> contribution stays a manual @Provides @IntoSet @KafkaConsumers binding.

Key Classes

`KafkaMessage<V>`

Immutable record handed to KafkaRecordHandler.handle.

public record KafkaMessage<V>(
        V value, String key, String topic, int partition,
        long offset, long timestamp, KafkaRecordHeaders headers) {}

key may be null; timestamp is epoch milliseconds. headers is the record’s headers as received — every header in order, with repeated keys, null values and binary values — and is immutable; it is never null (KafkaRecordHeaders.empty() for a record without headers).

// 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

There is no message.header(name). Use headers().lastHeader(name) for the header itself, or headers().asMap().get(name) for its text value.

`KafkaRecordContext`

The record’s metadata on the service side of a model-1, -2 or -3 dispatch: declare it as a handler parameter, or call KafkaRecordContext.current().

public record KafkaRecordContext(
        String consumerName, String topic, int partition, long offset, String key,
        long timestamp, String correlationId, KafkaRecordHeaders headers) {}

headers is the same immutable header collection the rest of the pipeline is given. As on KafkaMessage, there is no header(name): use headers().lastHeader(name) or headers().asMap().get(name). The context travels by reference over the local event bus only.

`KafkaRecordHeaders` and `KafkaRecordHeader`

The headers of one record exactly as they are on the wire. The consumer extracts them once per record, and that one immutable instance is what KafkaRecordFilter, deserializers, KafkaDispatchContext, KafkaRecordContext, KafkaMessage and KafkaConsumerRecordView all hold. Nothing a filter, deserializer, interceptor or handler does can change what another one reads.

package dev.vertique.kafka;

public record KafkaRecordHeader(String key, @Nullable io.vertx.core.buffer.Buffer value) {
    public static KafkaRecordHeader ofUtf8(String key, @Nullable String value);
    @Nullable public Buffer value();                       // a copy on every call
    public boolean hasValue();                             // no copy
    @Nullable public String valueAsUtf8();
    @Nullable public String valueAsString(Charset charset);
}

public record KafkaRecordHeaders(List<KafkaRecordHeader> entries)
        implements Iterable<KafkaRecordHeader> {
    public static KafkaRecordHeaders empty();
    public static KafkaRecordHeaders of(Map<String, String> textHeaders);
    public List<KafkaRecordHeader> headers(String key);
    public Optional<KafkaRecordHeader> lastHeader(String key);
    public Map<String, String> asMap();
}

Member	Behavior
`KafkaRecordHeader.value()`	The value as a Vert.x `Buffer`. `null` (no value), empty, and non-empty stay distinct. A `Buffer` is mutable, so the header copies it on construction and returns a new copy on every call: changing the buffer you passed in, or one you got back, never changes the header. Read it once into a local when you need it more than once
`KafkaRecordHeader.hasValue()`	`true` when the header carries a value, including an empty one; `false` when the value is `null`. It does not copy the buffer, so prefer it to `value() != null`
`KafkaRecordHeader.ofUtf8(key, text)`	A header whose value is the UTF-8 encoding of `text`; `null` text gives a header without a value
`valueAsUtf8()`, `valueAsString(Charset)`	Strict decoding: `null` for a `null` value, `""` for an empty one, `IllegalArgumentException` for bytes that are not valid in the charset
`KafkaRecordHeader.toString()`	Key and value length only — never the bytes
`KafkaRecordHeaders.of(Map<String, String>)`	One UTF-8 header per entry, in the map’s iteration order — pass a `LinkedHashMap` when order matters. A `null` map, key, or value throws `NullPointerException`; an empty map returns `empty()`
`entries()`, iteration	Every header in wire order. Duplicates are kept; keys are not trimmed or case-folded
`headers(key)`	Every header with that exact key, in order; empty list when absent
`lastHeader(key)`	The last header with that key. Empty only when the key is absent: a header present with a `null` value is returned
`asMap()`	Unmodifiable `Map<String, String>`. Lossy: a repeated key keeps its last non-`null` value, `null` values are left out, values are decoded as UTF-8 with malformed bytes replaced

KafkaRecordHeaders headers = KafkaRecordHeaders.of(Map.of("event-type", "order.created"));

Both records are immutable and compare by content; KafkaRecordHeaders equality is order-sensitive. asMap() builds a new map on every call, so call it once and keep the result when you read several keys.

`KafkaRecordFilter`

Functional interface applied before deserialization, so a rejected record never pays the deserialization cost. It sees the key and headers only.

boolean accept(String key, KafkaRecordHeaders headers);

headers is the record’s headers as received and is immutable: a filter can read every value of a repeated key, raw bytes and null values, and cannot add, remove or change a header.

KafkaRecordFilter signed = (key, headers) -> !headers.headers("signature").isEmpty();

The factories compare text — the last non-null value of the key, decoded as lenient UTF-8:

KafkaRecordFilter.headerEquals("event-type", "order.created")
KafkaRecordFilter.headerExists("x-correlation-id")
KafkaRecordFilter.headerMatches("version", Pattern.compile("v[12]"))
KafkaRecordFilter.headerIn("event-type", "order.created", "order.updated")
KafkaRecordFilter.allOf(filter1, filter2)
KafkaRecordFilter.anyOf(filter1, filter2)

headerExists is false when the key is absent or only has null values; a header with an empty value exists. headerEquals, headerMatches and headerIn reject such a record.

A filtered record is committed under MANUAL and reported to onRecordCompleted as SKIP. A filter that throws is handled like a deserialization failure: the record goes to the consumer’s error strategy and is reported to onRecordCompleted with that outcome.

`KafkaProducerFactory`

Creates typed producer proxies and exposes raw sends.

Method	Purpose
`create(Class<T> producerInterface)`	Builds the typed proxy for a `@KafkaProducer` interface
`send(String topic, String key, byte[] value, KafkaRecordHeaders headers)`	Raw send with pre-serialized bytes; the ambient durable context is captured and appended as context headers
`send(String topic, String key, byte[] value, KafkaRecordHeaders headers, DurableMetadata context)`	The same, with an explicit durable context instead of the ambient one
`sendForOutbox(String topic, String key, byte[] value, KafkaRecordHeaders headers, DurableMetadata context, String entryId)`	Raw send for one outbox entry, with an explicit durable context, tagged with `KafkaSendOrigin.OUTBOX` for capture hooks. The entry id is handed to the hooks as `KafkaProducerSend.originRef()`; it is not put on the record. `entryId` must not be `null`
`sendForOutbox(String topic, String key, byte[] value, KafkaRecordHeaders headers, DurableMetadata context)`	The same without an entry id: capture hooks see origin `OUTBOX` and a `null` `originRef`
`sendForDlq(String topic, String key, byte[] value, KafkaRecordHeaders headers)`	The framework’s dead-letter send, tagged with `KafkaSendOrigin.DLQ`. It forwards the given headers verbatim, including reserved `vertique-*` context headers, and adds no ambient context. The headers must contain `x-dlq-source-topic`, which the error handling always writes; without it the send fails with `IllegalArgumentException`. Not for application sends — use `send`
`close()`	Closes the underlying producer

Every send takes its headers as a KafkaRecordHeaders; null means no headers. See Headers on a sent record for what reaches the wire and which headers fail a send.

`KafkaConsumerDeploymentManager`

Method	Description
`deployAll()`	Registers codecs, then deploys every enabled consumer verticle in parallel; rolls back on partial failure
`undeployAll()`	Undeploys every consumer verticle

Typed Producer Proxies

Annotate an interface with @KafkaProducer; every method needs a @Topic.

@KafkaProducer
public interface OrderEvents {

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

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

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

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

@Provides @Singleton
static OrderEvents orderEvents(KafkaProducerFactory factory) {
    return factory.create(OrderEvents.class);
}

@KafkaProducer.name() defaults to "", which resolves to the interface’s simple name and is the key used for config overrides. A method has one of four parameter shapes:

Parameters	Interpretation
`(V value)`	value only; key is `null`
`(String key, V value)`	key and value
`(V value, KafkaRecordHeaders headers)`	value and headers
`(String key, V value, KafkaRecordHeaders headers)`	key, value, and headers

The shape is read from the parameter types. A last parameter of type KafkaRecordHeaders is the headers. Of the one or two parameters that remain, the last is the value and a leading String is the key. Two consequences:

(String value, KafkaRecordHeaders headers) is a String value with headers, not a key.
(String key, Map<String, String> value) is a key and a Map value.

factory.create(...) throws IllegalArgumentException, naming the method, for any other shape:

Rejected	Why
`(V value, Map headers)`, `(String key, V value, Map headers)`	A `Map` is not accepted as headers. Declare `KafkaRecordHeaders` and build it with `KafkaRecordHeaders.of(Map)`
A value of type `KafkaRecordHeaders`	`KafkaRecordHeaders` is only the last parameter, after the value
No parameters, more than three, headers before the value, or two leading parameters whose first is not a `String`	Not one of the four shapes

Method names must be unique within a @KafkaProducer interface. The topic and the serializer of a method are configured by its name, so factory.create(...) also throws IllegalArgumentException, naming the interface and both methods, when two methods share a name — for example send(Order) and send(Order, KafkaRecordHeaders). Give each its own name.

Headers on a sent record

Pass headers as a KafkaRecordHeaders. Build one from a text map, or from a list when you need a repeated key or a binary value:

// 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);

// No headers: pass null or KafkaRecordHeaders.empty().
orderEvents.orderCreatedWithHeaders(event, null);

The same rules hold for a proxy method and for send, the explicit-context send and sendForOutbox on KafkaProducerFactory:

Order on the wire. The record carries your headers first, in the order you gave them, with repeated keys and binary values unchanged. The framework then appends one text header per durable-context namespace (vertique-<namespace>). The order among those context headers is unspecified.
Reserved prefix. A header whose key starts with vertique- fails the send: the returned future fails with IllegalArgumentException naming the key, and nothing is sent.
Null values. A header with a null value fails the send the same way. Kafka allows such a header, but the producer record cannot carry it. An empty value is fine.
The serializer sees your headers. KafkaSerializer.serialize is given the headers you passed (never null — KafkaRecordHeaders.empty() when you passed none), before the context headers are appended. They are immutable, so a serializer cannot add a header to the record.

sendForDlq is the exception: it forwards the headers it is given verbatim, reserved-prefix headers included, and appends no context. A null-valued header fails it too.

Topic, format, serde properties, and JSON mapper profile are overridable per method at kafka.producers.{producerName}.methods.{methodName}.*.

To pick a JSON mapper profile, place @JsonProfile("...") at type level on the interface. It is a separate annotation, not a @KafkaProducer attribute, and a method-level placement is rejected at build time.

Configuration

{
  "kafka": {
    "bootstrap.servers": "localhost:9092",
    "security.protocol": "SASL_SSL",
    "sasl.mechanism": "PLAIN",
    "sasl.username": "user",
    "sasl.password": "secret",
    "format": "json",
    "jsonProfile": "cms",
    "properties": { "auto.offset.reset": "earliest" },
    "schemaRegistry": { "url": "http://schema-registry:8080/apis/registry/v3" },
    "messageKeyHash": { "hmacSecret": "…" },
    "consumers": {
      "order-created": {
        "topic": "order.created",
        "groupId": "order-processing",
        "enabled": true,
        "commitStrategy": "MANUAL",
        "errorStrategy": "DEAD_LETTER",
        "deadLetterTopic": "order.created.dlq",
        "eventBusTimeoutMs": 30000,
        "maxInFlight": 256,
        "instances": 2,
        "worker": false,
        "format": "avro",
        "jsonProfile": "cms",
        "retry": {
          "maxRetries": 3,
          "backoffMs": 1000,
          "backoffMultiplier": 2.0,
          "maxBackoffMs": 60000,
          "exhaustedStrategy": "DEAD_LETTER"
        },
        "serdeProperties": { "apicurio.registry.auto-register": false },
        "properties": { "max.poll.records": "100" }
      }
    },
    "producer": { "properties": { "acks": "all" } },
    "producers": {
      "OrderEvents": {
        "format": "avro",
        "jsonProfile": "cms",
        "methods": {
          "orderCreated": {
            "topic": "order.created.override",
            "jsonProfile": "strict",
            "serdeProperties": { "apicurio.registry.auto-register": false }
          }
        }
      }
    }
  }
}

`kafka.consumers.{name}.*`

Field	Default	Constraint / description
`topic`	annotation or builder value	Topic to consume
`groupId`	annotation or builder value	Consumer group id
`enabled`	`true`	Disabled consumers are not deployed
`commitStrategy`	annotation value (`AUTO`)	`AUTO` or `MANUAL`
`errorStrategy`	annotation value (`SKIP`)	`SKIP`, `DEAD_LETTER`, or `RETRY`
`deadLetterTopic`	`"{topic}.dlq"`	Used by `DEAD_LETTER`
`eventBusTimeoutMs`	`30000`	Must be `> 0`
`maxInFlight`	`256`	Must be `>= 1`. The consumer pauses at this many in-flight dispatches and resumes as the queue drains
`instances`	`1`	Must be `>= 1`
`worker`	unset	Event-loop threading unless a `mayBlock()` serde forces worker; see below
`format`	per-endpoint, else `kafka.format`, else auto-detect, else `json`	Any registered provider format
`jsonProfile`	unset	JSON mapper profile id for value deserialization
`serdeProperties`	—	Serde and registry config; never passed to the Kafka client
`properties`	—	Kafka-client property overrides for this consumer only
`retry.maxRetries`	`3`	Must be `>= 0`
`retry.backoffMs`	`1000`	Must be `>= 0`
`retry.backoffMultiplier`	`2.0`	Must be `>= 1.0`
`retry.maxBackoffMs`	`60000`	Must be `>= 0`
`retry.exhaustedStrategy`	`DEAD_LETTER`	Must not be `RETRY`

A violated constraint fails startup with a ConfigurationException naming the exact config path.

Kafka client property merge order

Later entries win:

root-level connection scalars — bootstrap.servers, security.protocol, sasl.mechanism, sasl.jaas.config;
global kafka.properties;
per-consumer kafka.consumers.{name}.properties.

sasl.jaas.config is constructed automatically when sasl.username and sasl.password are present and sasl.jaas.config is absent.

Producers merge fixed serializers, then the root-level connection scalars, then kafka.properties, then kafka.producer.properties. The framework sets no acks or enable.idempotence, so producers inherit the Kafka client defaults (acks=all and enable.idempotence=true with the kafka-clients 4.x line brought in by Vert.x) unless one of those property bags sets them.

When bootstrap.servers is not configured through any source, producer startup fails with an IllegalStateException rather than silently falling back to localhost:9092.

Properties files. With Vert.x hierarchical expansion (hierarchical=true), dotted keys such as kafka.bootstrap.servers expand into nested JSON. Both the flat and nested forms resolve.

`serdeProperties` versus `properties`

properties is passed verbatim to the Vert.x Kafka client and must contain only Kafka-client keys. Serde and schema-registry configuration lives in its own namespace:

Path	Purpose
`kafka.schemaRegistry.*`	Global schema-registry config (URL, auth)
`kafka.consumers.{name}.serdeProperties.*`	Per-consumer serde overrides
`kafka.producers.{name}.serdeProperties.*`	Per-producer serde overrides
`kafka.producers.{name}.methods.{method}.serdeProperties.*`	Per-method serde overrides

Merge order, most specific last: kafka.schemaRegistry.* → per-producer or per-consumer serdeProperties → per-method serdeProperties.

Secrets

Secret keys inside properties, serdeProperties, schemaRegistry, and the root connection properties are masked when a config object is rendered, so a startup log or a ConfigurationException never reveals a credential. kafka.messageKeyHash.hmacSecret is write-only — it deserializes in and is never serialized back out.

JSON mapper profile precedence

Consumers, highest first:

kafka.consumers.{name}.jsonProfile
@JsonProfile("...") at type level on the @KafkaListener type, or KafkaConsumerBinding.Builder#jsonProfile(...)
kafka.jsonProfile
json.jsonProfile
the vertique framework default, resolved through the registry

Producers, highest first:

kafka.producers.{name}.methods.{method}.jsonProfile
kafka.producers.{name}.jsonProfile
@JsonProfile("...") at type level on the @KafkaProducer type
kafka.jsonProfile
json.jsonProfile
the vertique framework default

A custom KafkaDeserializer set through KafkaConsumerBinding.Builder#deserializer() bypasses profile selection entirely — the profile applies only when the framework builds the serde from a KafkaSerdeProvider. A method-level @JsonProfile on a Kafka listener or producer type is rejected at build time.

Error Handling

`ErrorStrategy`

Value	Requires	Behavior
`SKIP`	any commit strategy	Log the error and commit, dropping the record
`DEAD_LETTER`	`MANUAL`	Publish the original bytes and headers to the DLQ topic, then commit. If the DLQ publish fails the offset is not committed and Kafka redelivers
`RETRY`	`MANUAL`	Pause the consumer for a backoff, then resume; Kafka redelivers the uncommitted record. After `maxRetries` redeliveries, fall back to `exhaustedStrategy`

How `RETRY` works

RETRY uses Kafka’s native redelivery, not an application-level retry loop. On failure the offset is not committed, the consumer pauses, and after the backoff it resumes and Kafka redelivers from the uncommitted offset. The record then re-enters the full pipeline, including any service-level @Retry, @Timeout, and @CircuitBreaker policies.

Retry delays use the shared vertique-resilience exponential backoff policy with zero jitter. The configured backoffMs, backoffMultiplier, and maxBackoffMs therefore retain deterministic Kafka timing while using the canonical retry-count validation and capped delay calculation.

Record B fails (attempt 1/3)
  → offset not committed
  → pause consumer, resume in 1s
  → Kafka redelivers B, full pipeline re-runs
  → fails again (2/3) → backoff 2s
  → fails again (3/3) → backoff 4s
  → fails again (exhausted):
      exhaustedStrategy = DEAD_LETTER → publish to DLQ, commit
      exhaustedStrategy = SKIP        → log, commit

Kafka RETRY and a service operation’s @Retry compose multiplicatively. A service with @Retry(maxRetries = 3) behind a consumer with RETRY(maxRetries = 3) gives a persistently failing record up to twelve dispatch attempts before the exhausted strategy applies.

RETRY blocks the partition. No other record from that partition is delivered during the backoff. That is appropriate for failures measured in seconds; for failures measured in hours use DEAD_LETTER and reprocess from the DLQ.

Poison-message guarantee

A record that always fails leaves the stream after maxRetries attempts. The longest it can block a partition is:

sum(backoffMs × backoffMultiplier^i for i in 0..maxRetries-1), each term capped at maxBackoffMs

With the defaults (3 retries, 1s base, ×2, 60s cap) the worst case is 1s + 2s + 4s = 7 seconds.

RetryConfig.DEFAULT — maxRetries = 3, backoffMs = 1000, backoffMultiplier = 2.0, exhaustedStrategy = DEAD_LETTER, maxBackoffMs = 60000 — applies when errorStrategy = RETRY and no retry block is configured.

Value Formats

`KafkaSerdeProvider`

The pluggable value-format SPI, contributed via @IntoSet Set<KafkaSerdeProvider>. KafkaModule declares the multibinding but contributes no provider.

public interface KafkaSerdeProvider {

    String format();                                        // e.g. "json", "avro"

    <V> KafkaSerializer<V> serializer(Class<V> type, JsonObject endpointConfig);

    <V> KafkaDeserializer<V> deserializer(Class<V> type, JsonObject endpointConfig);

    default boolean autoDetects(Class<?> type) { return false; }

    default boolean supports(Class<?> type) { return true; }

    /** {@code true} when this format's serdes may block (e.g. a remote schema registry). */
    default boolean mayBlock() { return false; }

    /** Property-based routing for model-3 routers. Throws UnsupportedOperationException by default. */
    default KafkaDeserializer<Object> routingDeserializer(JsonObject endpointConfig) { … }

    /** Extracts the match value for a property route. Throws UnsupportedOperationException by default. */
    default String matchValue(Object deserializedValue, String property) { … }

    /** Converts the routing value to the matched route's target type. */
    default <V> V convertRouted(Object routingValue, Class<V> type, JsonObject endpointConfig) { … }
}

routingDeserializer is built once per router consumer from the merged endpoint config and is type-agnostic, so the format decodes the record before the route is chosen. convertRouted then turns that value into the matched route’s type; the default implementation reuses it when it is already an instance of the target type and throws DeserializationException otherwise. A format that decodes an intermediate representation while routing should override it — the JSON provider does, calling treeToValue on the already-parsed tree so there is no second parse.

public interface KafkaSerializer<V> {
    byte[] serialize(V value, String topic, KafkaRecordHeaders headers);
    default boolean mayBlock() { return false; }
    default void close() {}
}

public interface KafkaDeserializer<V> {
    V deserialize(byte[] data, String topic, KafkaRecordHeaders headers) throws DeserializationException;
    default boolean mayBlock() { return false; }
    default void close() {}
}

Both carry topic and headers because schema-registry-backed formats need them for subject naming. A serializer is given the application headers of the record being sent as an immutable KafkaRecordHeaders — never null, and without the framework context headers, which are appended after serialization. A deserializer is given the record’s headers as received, as the same immutable KafkaRecordHeaders the filter and the handler get — never null. It can read them, for example to pick a format from a content-type header, and cannot change them.

Format precedence

First match wins:

the endpoint’s own format — consumer or producer-method config;
global kafka.format;
auto-detection — each registered provider’s autoDetects(valueType) in registration order;
"json", the terminal fallback key.

"json" is an ordinary provider key, not a built-in. Resolving to it still requires a registered provider, so a format with no provider fails startup with an IllegalArgumentException naming the missing module.

Worker threading for blocking formats

When the effective deserializer’s mayBlock() is true:

the consumer verticle is forced to worker threading;
an explicit worker = false is rejected at startup;
an absent worker field is silently forced to true.

This contract, not a timeout, is what keeps a remote registry call off the event loop. KafkaSerdeRegistry wraps every provider-built serde so mayBlock() delegates to the provider — a provider cannot forget to stamp it.

When a provider is required

An endpoint needs a registered provider for its resolved format whenever the framework must build a serde or routing artifact — including for "json". Paths that build nothing need no provider.

Path	Provider required
Enabled binding or handler with a framework-built deserializer	Yes
Disabled binding or handler with a declared `valueType`	Yes
Router with any property route or a non-`Void` payload route	Yes
All-`Void` property router (the selector is still read)	Yes
Model-2 binding with a custom `deserializer`	No
Model-4 raw `KafkaRecordHandler`	No
Non-property `Void` or no-payload binding or handler	No
Header- or default-only all-`Void` router	No

Adding a format

@Provides @Singleton @IntoSet
static KafkaSerdeProvider myFormatProvider(@VertxConfig JsonObject config) {
    return new MyFormatSerdeProvider(config);
}

See dev.vertique:vertique-kafka-avro for a complete reference implementation.

Extension Points

`KafkaConsumerInterceptor` (multibinding)

KafkaConsumerInterceptor extends OrderedExtension and is sorted by OrderedExtension.comparator() — phase, then ascending priority, then orderKey() (the FQCN by default). Lower priority runs first.

Sync observers are fire-and-forget. An Exception, LinkageError, AssertionError or StackOverflowError thrown by one is logged and swallowed; later interceptors still run and the record’s outcome is unaffected. An Exception, AssertionError or StackOverflowError is logged at WARN each time, naming the interceptor class, the callback and the failure’s class only; the failure itself, with its message, is logged at DEBUG. A LinkageError means the callback cannot run at all, so it is logged at ERROR at a limited rate per interceptor class and callback — the first time, then at most once every five minutes with the number of failures in between — saying that the callback is unusable and its notifications are being lost.

Callback	When
`void onRecord(KafkaDispatchContext<?> ctx)`	After deserialization, before `beforeDispatch`
`void onSuccess(KafkaDispatchContext<?> ctx)`	After a successful dispatch
`void onError(KafkaDispatchContext<?> ctx, Throwable error)`	On any dispatch failure
`void onRecordCompleted(KafkaConsumerCompletedEvent event, KafkaConsumerRecordView record)`	Once per received record, when its disposition is final — see Record completion

Async handlers can change the outcome.

Callback	Failure behavior
`Future<KafkaDispatchContext<?>> beforeDispatch(KafkaDispatchContext<?> ctx)`	A failed future short-circuits the dispatch. Return the context — this is how `withFiltered(true)` and added attributes propagate
`Future<Void> afterDispatch(KafkaDispatchContext<?> ctx)`	Failures are logged, not propagated
`Future<Void> recoverError(KafkaDispatchContext<?> ctx, Throwable error)`	A succeeded future means handled — the record is committed. A failed future means proceed with the configured error strategy

@Override
public Future<Void> recoverError(KafkaDispatchContext<?> ctx, Throwable error) {
    if (error instanceof BeanValidationException) {
        return Future.succeededFuture();   // handled — commit the record
    }
    return Future.failedFuture(error);     // unhandled — apply SKIP / DEAD_LETTER / RETRY
}

Interceptors are offered recoverError in extension order and the first succeeded future wins; later interceptors are not invoked.

KafkaDispatchContext<V> is an immutable record — (consumerName, topic, partition, offset, key, value, rawEvidence, headers, timestamp, retryCount, filtered, attributes) — with copy-on-write withFiltered(boolean) and withAttribute(String, Object). headers is the record’s KafkaRecordHeaders as received; both with… methods carry the same instance over. retryCount is 0-based and counts Kafka-native redeliveries, so a retry-topic routing decision in recoverError can read it.

Record completion: `onRecordCompleted`

KafkaConsumerInterceptor.onRecordCompleted is the one place that sees every record the consumer received, with its final outcome. It is a sync observer like the three above, on the same interceptor multibinding and in the same order.

public interface KafkaConsumerInterceptor extends OrderedExtension {
    default void onRecordCompleted(
            KafkaConsumerCompletedEvent event, KafkaConsumerRecordView record) {}
}

public record KafkaConsumerCompletedEvent(
        KafkaConsumerRecordIdentity identity, KafkaTerminalOutcome outcome) {}

public record KafkaConsumerRecordIdentity(
        String consumerName, String topic, int partition, long offset,
        long timestamp, int retryCount) {}

public interface KafkaConsumerRecordView {
    KafkaConsumerRecordIdentity identity();
    @Nullable String key();
    KafkaRecordHeaders headers();    // every header as received: ordered, duplicates, raw bytes
    PayloadSource value();           // the broker's array, NOT copied; absent for a tombstone
}

When it runs. Exactly once per delivered record, when the disposition is final and all async work — DLQ publish, seek, recovery, afterDispatch — has settled. It runs on every path: filtered before deserialization, rejected by a filter that throws, no matching route, deserialization failure, filtered or failed in beforeDispatch, dispatched, recovered, or handed to the error strategy. A record redelivered for a retry completes once per delivery; retryCount is the number of retries before that attempt. The one limit: an asynchronous interceptor or handler future that never completes, or a throw from inside an asynchronous continuation, leaves the record without a completion.

Observer only. The decision is already made. The callback cannot change filtering, dispatch, commit, retry, or dead-lettering.

Threading. onRecordCompleted may run on any thread, and concurrently for different records. It runs on whichever thread settles the record’s last future: the consumer’s context, a handler’s or interceptor’s thread, or the producer’s context on a dead-letter path. Implementations must be thread-safe.

The event is facts only. KafkaConsumerCompletedEvent has no key, headers, value, or dispatch context, so it can be logged or passed to a metrics observer as is.

The view is framework-owned. The consumer builds it once per record, before the KafkaRecordFilter, route resolution, deserialization, and the interceptor chain:

identity(), key() and headers() are the record’s own. A beforeDispatch that returns a different KafkaDispatchContext changes none of them, and the headers are immutable, so no filter or deserializer can edit them.
event.identity() and record.identity() are the same instance.
A consumer with no interceptors builds no view. The headers are still extracted once per record, because the filter, the deserializers and the handler are given them.

headers() is the wire form. It returns a KafkaRecordHeaders: every header in the order the broker delivered it, with repeated keys, null values, and the exact value bytes. It is an immutable snapshot. See KafkaRecordHeaders for the lookups.

KafkaRecordHeaders headers = record.headers();

// The last header for a key. Empty only when the key is absent.
String traceId = headers.lastHeader("trace-id")
        .map(KafkaRecordHeader::valueAsUtf8)   // null for a null value; throws on invalid UTF-8
        .orElse(null);

// Every header for a repeated key, in order, as raw bytes.
for (KafkaRecordHeader hop : headers.headers("x-forwarded-by")) {
    Buffer raw = hop.value();                  // a copy; null when the header has no value
}

// The text map that filters, deserializers and handlers receive. Lossy.
Map<String, String> text = headers.asMap();

asMap() is a lossy text projection for code that needs a Map<String, String>: a repeated key keeps only its last non-null value, a header with a null value is left out, and every value is decoded as UTF-8 with malformed bytes replaced, so a binary value cannot be recovered from it.

value() is not a snapshot. It is the array the broker delivered, uncopied. The deserializer, router property matching and — for a byte[] consumer — the handler were given the same array, so an in-place edit by any of them is visible through value(). The framework never copies a record value. Copy the bytes before retaining them past the callback, and do not treat them as proof of the wire content when application code may edit the array. KafkaDispatchContext.rawEvidence() is the same kind of no-copy view.

Need	Callback
The deserialized value of records that reached dispatch	`onRecord`
Every record, its outcome, and its raw form (key, headers, value bytes)	`onRecordCompleted`

onRecord never sees a record that was filtered before deserialization, matched no route, or failed to deserialize, and it runs before the outcome is known.

@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());
    }
}

`KafkaTerminalOutcome`	Meaning
`SUCCESS`	The handler completed successfully
`SKIP`	Skipped per `ErrorStrategy.SKIP`, filtered, or no matching route
`RECOVERED`	An interceptor’s `recoverError` handled the error
`RETRY_SCHEDULED`	Consumer paused, offset not committed, Kafka will redeliver
`DLQ_PUBLISHED`	Forwarded to the dead-letter topic
`DLQ_FAILED`	The DLQ publish failed; the offset was not committed
`ERROR_HANDLER_FAILED`	The error handler’s own future failed; the record’s disposition is unknown

Breaking change (core 0.3.0). The separate consumer capture-hook SPI, its terminal and raw-disposition records, and its Set multibinding are removed. Implement KafkaConsumerInterceptor.onRecordCompleted and contribute the interceptor into Set<KafkaConsumerInterceptor> instead. Read coordinates and retryCount from event.identity(), and the key, headers and raw value from the KafkaConsumerRecordView; record.headers().asMap() gives the headers as a Map<String, String>. The dispatch context is no longer delivered at completion. The record value is no longer cloned for observers. onRecord, onSuccess and onError now also swallow AssertionError and LinkageError.

`KafkaProducerCaptureHook` (multibinding)

Fires exactly once per send, through the shared wire funnel in KafkaProducerFactory, after producer.send(record) settles. Also extends OrderedExtension and observer-only. An Exception, LinkageError, AssertionError or StackOverflowError thrown by onSend is logged and swallowed; later hooks still run and the send’s result is unaffected. An Exception, AssertionError or StackOverflowError is logged at WARN each time, by class name, with the failure itself at DEBUG, and a LinkageError at ERROR at a limited rate per hook class — the first time, then at most once every five minutes — saying that the hook is unusable and its notifications are being lost.

The framework calls onSend(KafkaProducerSend send). Override this form: new send details are added to the record as components, never as further positional parameters. Read type-level annotations from operation.producerType() — the @KafkaProducer interface the application injected — not from method.getDeclaringClass(), which is the super-interface when the send method is inherited.

KafkaProducerSend components:

Component	Notes
`origin`	`KafkaSendOrigin` — `DIRECT_PRODUCER`, `OUTBOX`, `DLQ`, or `INTERNAL`
`topic`	Target topic
`key`	Record key, or `null`
`value`	No-copy `PayloadSource` over the serialized wire bytes
`headers`	The `KafkaRecordHeaders` of the record as sent, in wire order: the application headers, then the framework context headers. Repeated keys and binary values are kept; `headers.asMap()` gives a lossy text map
`operation`	A `KafkaProducerOperation(producerType, producerName, method)`; non-`null` only for `DIRECT_PRODUCER`. `operation().method()` is how a hook reaches method-level annotations on the direct-producer path
`result`	The settled `AsyncResult<RecordMetadata>`
`originRef`	The outbox entry id, as a string, for an `OUTBOX` send made through the `sendForOutbox` form that takes one; `null` for every other send. A hook uses it to join the wire bytes to the outbox entry they belong to

originRef is not authenticated. Any code that holds the KafkaProducerFactory can call sendForOutbox with origin OUTBOX and an arbitrary entry id, so the reference alone does not prove that the relay sent the record. A consumer of the hook that joins on it must also match a relay notification (OutboxPublishObserver) for the same entry id and destination.

originRef is the last component. The seven-argument KafkaProducerSend constructor without it remains and gives a null originRef.

Deprecated: the positional onSend. The seven-argument form onSend(origin, topic, key, value, headers, producerMethod, result) is @Deprecated. It cannot carry originRef or any component added later; onSend(KafkaProducerSend) replaces it. It keeps working: the default onSend(KafkaProducerSend) still delegates to it, passing operation().method() as producerMethod, so a hook that overrides only the positional form is still called for every send. Never make the positional form delegate back to the event form, which would recurse.

The protected KafkaProducerFactory.sendWire/fireHooks funnel carries the KafkaProducerOperation and, in the overloads with a trailing originRef parameter, the origin reference. A send without an origin reference goes through the sendWire form without that parameter, which delegates to the one with it; a send with a reference calls the form with the parameter directly. fireHooks follows the same rule. A subclass that intercepts every send overrides both, or only the form with originRef. A subclass that overrides only the older sendWire/fireHooks forms, the ones without originRef, does not see sends that carry an origin reference.

KafkaProducerFactory.create(Class) also calls validateProducer(Class<?> producerInterface) on every hook, once, before it builds the proxy. Unlike onSend, it may throw: the exception propagates out of create, so a configuration error a hook can detect statically (such as an unknown audit policy id on a producer method) fails application startup instead of surfacing, or being swallowed, on a later send. The default accepts every producer interface.

Other multibindings

Multibinding	Qualifier	Contributes
`Set<KafkaConsumerBinding<?>>`	—	Model-2 declarative bindings
`Set<Object>`	`@KafkaConsumers`	Model-3 router `Class<?>` literals and model-4 handler instances
`Set<KafkaConsumerInterceptor>`	—	Consumer pipeline interceptors, including record-completion observers
`Set<KafkaProducerCaptureHook>`	—	Producer boundary capture hooks
`Set<KafkaSerdeProvider>`	—	Value-format providers

Module Dagger Bindings

KafkaModule transitively includes ContextRuntimeModule, LoggingContextModule, DispatchModule, DeployerModule, and HealthCheckModule.

Type	Scope	Description
`KafkaConfig`	`@Singleton`	Typed `kafka` config parsed at the module boundary
`KafkaConsumerRegistry`	`@Singleton`	Registry built from every consumer source, validated at startup
`KafkaConsumerDeploymentManager`	`@Singleton`	Deploys and undeploys consumer verticles
`KafkaProducerFactory`	`@Singleton`	Typed producer proxies and raw sends
`HealthCheck` (`@IntoSet`)	—	`KafkaConsumerHealthCheck`, contributed to the readiness set

Failures, Constraints, and Common Mistakes

Startup failures

Failure	Cause
`ConfigurationException`	A consumer config bound is violated — an unknown `commitStrategy`, `errorStrategy`, or `retry.exhaustedStrategy`, `eventBusTimeoutMs <= 0`, `maxInFlight < 1`, `instances < 1`, `maxRetries < 0`, `backoffMs < 0`, `backoffMultiplier < 1.0`, `maxBackoffMs < 0`. Enum failures identify the invalid type and value; other validation messages name the exact config path
`KafkaRegistrationException`	An invalid consumer declaration, or a generated binding companion that is present but malformed. A `ConfigurationException` subtype — broken generated code is never silently skipped
`IllegalArgumentException`	The resolved format has no registered `KafkaSerdeProvider`; the message names the missing module
`IllegalStateException`	`worker = false` on a consumer whose effective deserializer reports `mayBlock()`
`IllegalStateException`	`bootstrap.servers` missing for a producer

Runtime failures

Failure	Meaning
`DeserializationException`	The value could not be decoded. A `TechnicalException` subtype; it enters the configured error strategy like any other dispatch failure
Event-bus timeout	Dispatch exceeded `eventBusTimeoutMs`; treated as a dispatch failure
Failed send future with `IllegalArgumentException`	A header on a producer send uses the reserved `vertique-` prefix or has a `null` value. The message names the key; nothing was sent

Common mistakes

AUTO commit with DEAD_LETTER or RETRY. Both strategies depend on withholding the commit; AUTO has already committed. Use MANUAL.
Non-idempotent handlers under MANUAL. At-least-once means redelivery is normal, not exceptional.
Using RETRY for a long outage. The retrying partition is blocked for the whole backoff. Use DEAD_LETTER and reprocess.
Dropping the context returned by beforeDispatch. KafkaDispatchContext is immutable; withFiltered(true) has no effect unless the returned future carries the new instance.
Retaining a PayloadSource across threads. It is a no-copy view of the record buffer — copy first.
Contributing a model-3 router instance. A routing interface contributes its Class<?> literal; models 1 and 4 contribute instances.
Putting serde settings in properties. That bag goes verbatim to the Kafka client. Serde and registry settings belong in serdeProperties and kafka.schemaRegistry.
Expecting consumers to start themselves. Nothing calls deployAll() for you.
Installing KafkaModule alone. With no format provider, any consumer needing a serde fails at startup.
Declaring a Map header parameter on a @KafkaProducer method. factory.create(...) rejects it. Declare KafkaRecordHeaders and pass KafkaRecordHeaders.of(map).
Looking for a single-header accessor on KafkaMessage or KafkaRecordContext. There is none. Use headers().lastHeader(name) for the header as received, or headers().asMap().get(name) for its text value.
Trying to add or change a header in a filter or deserializer. The headers are immutable; entries() and asMap() reject changes, and a value Buffer is a copy.
Calling asMap() for every key. Each call builds a new map. Keep the result.
Setting a vertique-* header yourself. The prefix belongs to the framework; the send fails.

Dependencies

Artifact	Scope	Purpose
`dev.vertique:vertique-core`	compile	DI wiring, typed config parsing, `HealthCheck` SPI, exception roots
`dev.vertique:vertique-context`	compile	Dispatch-context propagation across the event bus
`dev.vertique:vertique-logging`	compile	MDC binding for the record’s correlation id
`dev.vertique:vertique-services`	compile	Service contract registry and target resolution for dispatch
`dev.vertique:vertique-deploy`	compile	Verticle deployment and lifecycle phases
`io.vertx:vertx-kafka-client`	compile	Vert.x Kafka consumer and producer
`com.fasterxml.jackson.core:jackson-annotations`	compile	Annotations only, for the typed config records

vertique-kafka-core carries no Jackson databind, Avro, or Apicurio dependency at compile scope — actual value serialization lives in the format modules. vertique-kafka-json brings jackson-databind; vertique-kafka-avro brings org.apache.avro:avro and the Apicurio Avro serde.