Kafka Avro

Status: Stable Package: dev.vertique.kafka.avro Artifact: vertique-kafka-avro Depends on: kafka-core API reference: javadoc.io

Avro value-format support for the vertique-kafka family using Apicurio Registry as the Schema Registry. Adds the "avro" format to the pluggable KafkaSerdeProvider SPI declared in vertique-kafka-core. One of two sibling format modules under the vertique-kafka aggregator (vertique-kafka-json provides JSON; this module provides Avro). Both are opt-in — the application composes the format module(s) it needs.

---

When To Use It

Add this module when:

• Producers or consumers exchange Avro-encoded SpecificRecord payloads with a Schema Registry.
• Topics are governed by schemas managed in Apicurio Registry (or any registry that speaks the Confluent-compatible wire format).
• Auto-detection of SpecificRecord value types is desired without configuring an explicit format = avro.

Do not use when only JSON payloads are needed — add vertique-kafka-json (and KafkaJsonModule) instead.

---

Core Concepts

Opt-in by module composition

vertique-kafka-avro contributes a single KafkaSerdeProvider into the Set<KafkaSerdeProvider> multibinding declared by KafkaModule in vertique-kafka-core. Including AvroModule in the Dagger component is the only activation step. Excluding it means Avro is unavailable; any format resolution that reaches "avro" fails fast at startup with an actionable error.

Byte-level wrapping, Confluent-wire format

Avro serialization operates at the byte[] level inside the framework’s own KafkaSerializer/KafkaDeserializer wrappers. The native Kafka client stays on ByteArraySerializer/ByteArrayDeserializer. The wire encoding is Confluent-wire-compatible: magic byte + 4-byte schema-id prepended to the payload, Kafka headers disabled. This format survives the consumer’s UTF-8 header stringification and the DLQ raw-bytes republish intact.

SpecificRecord only

The first-class payload model is org.apache.avro.specific.SpecificRecord (classes generated by avro-maven-plugin with specific.avro.reader=true). GenericRecord and reflection-based payloads are not supported.

Avro class trust (Avro 1.12.2+)

Avro 1.12.2 checks every class it resolves from a schema name — records, enums and fixed types, on both the read and the write path — against one JVM-wide allow-list. Generated SpecificRecord classes are not on it by default, so the provider extends the list itself when a serde is built:

• Declared types — serializer(type, …) and deserializer(type, …) trust type plus every nested record, enum and fixed class reachable from its schema, as exact class names. No configuration is needed for typed producers, consumers and router routes.
• Router routing — the type-agnostic routingDeserializer receives no types. Route types are built first, so declared route types are already trusted; list vertique.avro.trusted-packages for any record that is only reachable through the wire schema (for example a writer schema that names a class no route declares).
• Additive — the provider composes onto the installed global predicate and never replaces it, so org.apache.avro.SERIALIZABLE_CLASSES / SERIALIZABLE_PACKAGES and any predicate your application installed keep working. If another component replaces the global, the next serde build restores the provider’s trust.
• JVM-wide — trust registered for one endpoint is visible to every Avro resolution in the process. It is registered when serdes are built at startup, never per record.
• Fail-closed — a writer schema naming a class that was neither declared nor listed is rejected, and deserialization surfaces it as a DeserializationException; a producer surfaces the SecurityException.

Event-loop safety

ApicurioAvroSerdeProvider.mayBlock() returns true. The KafkaSerdeRegistry propagates this flag onto every serde it builds. Consequently:

• Producers: serialization is offloaded to a worker thread via vertx.executeBlocking on each send.
• Consumers: the worker threading model is forced for any consumer using the "avro" format. An explicit worker = false in consumer config is rejected at startup; an absent worker field is forced to true.

Serdes are built once per endpoint

serializer(...) and deserializer(...) are called once per producer-method or consumer at startup. Per-record serde construction does not occur; the Apicurio registry client’s schema cache is active by default.

Configuration is separated from native Kafka properties

The native properties bag is passed verbatim to the Vert.x Kafka client and must contain only Kafka-client keys. Serde and registry configuration lives in the dedicated serdeProperties namespace and the kafka.schemaRegistry.* block; ApicurioAvroSerdeProvider never reads the native properties bag.

---

Key Classes

ApicurioAvroSerdeProvider

KafkaSerdeProvider implementation for the "avro" format. Backed by io.apicurio:apicurio-registry-avro-serde-kafka:3.3.3.

public final class ApicurioAvroSerdeProvider implements KafkaSerdeProvider {

    public static final String FORMAT = "avro";

    /** Constructed by AvroModule from the typed Kafka config; reads its schemaRegistry block. */
    public ApicurioAvroSerdeProvider(KafkaConfig kafkaConfig) { ... }

    @Override public String format() { return FORMAT; }

    /** Returns true for SpecificRecord subtypes — triggers auto-detection. */
    @Override public boolean autoDetects(Class<?> type) { ... }

    /** Always true — Schema Registry calls may block on a cache miss. */
    @Override public boolean mayBlock() { return true; }

    /**
     * Builds a per-endpoint AvroKafkaSerializer. The returned serializer implements
     * {@code serialize(V value, String topic, KafkaRecordHeaders headers)}: it uses the topic for
     * the registry subject and ignores the headers, since the schema id travels in the payload.
     */
    @Override public <V> KafkaSerializer<V> serializer(Class<V> type, JsonObject endpointConfig) { ... }

    /**
     * Builds a per-endpoint AvroKafkaDeserializer. The returned deserializer implements
     * {@code deserialize(byte[] data, String topic, KafkaRecordHeaders headers)}: it uses the topic
     * for the registry lookup and ignores the headers, since the schema id travels in the payload.
     */
    @Override public <V> KafkaDeserializer<V> deserializer(Class<V> type, JsonObject endpointConfig) { ... }

    /**
     * Type-agnostic deserializer for Model-3 router property routing, built from the endpoint
     * config. It has the same {@code deserialize} signature and ignores the headers too.
     */
    @Override public KafkaDeserializer<Object> routingDeserializer(JsonObject endpointConfig) { ... }

    /** Reads a discriminator field from a GenericRecord for Model-3 matchProperty routing. */
    @Override public String matchValue(Object deserializedValue, String property) { ... }
}

Config resolution inside ApicurioAvroSerdeProvider:

The merged endpointConfig passed by the framework already combines serdeProperties + schemaRegistry in the correct precedence order. The provider reads:

Key path in endpointConfig	Apicurio config key set	Notes
schemaRegistry.url	apicurio.registry.url	Schema Registry URL
serdeProperties.*	Any apicurio.registry.* key	User properties overlay, highest priority
serdeProperties."vertique.avro.trusted-packages"	— (read by the provider, never forwarded to Apicurio)	Extra trusted package prefixes; see Avro class trust

Fixed defaults applied by the provider (not overridable via serdeProperties without explicit re-setting):

Apicurio key	Serializer	Deserializer	Value
apicurio.registry.headers.enabled	yes	yes	false (Confluent-wire: schema-id in payload, not headers)
apicurio.registry.use-specific-avro-reader	—	yes	true
apicurio.registry.auto-register	yes	—	true (dev convenience; override via serdeProperties)
apicurio.registry.retry-count	yes	yes	3 (bounded retries; override via serdeProperties)
apicurio.registry.retry-backoff-ms	yes	yes	300 (override via serdeProperties)

Registry-call bounding (NFR-AVRO-002): the JDK client adapter applies a 30s connect timeout and the retry knobs above bound retry storms; the worker-threading contract keeps blocking calls off the event loop. Apicurio 3.3.3 has no config-map key for a per-request socket read timeout (a custom registry client would be needed — see the PRD’s Deferred items).

Invariants and Gotchas

• The router-routing deserializer (routingDeserializer(endpointConfig)) is built once per router consumer (held by the dispatcher) from the consumer’s merged endpointConfig, so endpoint serdeProperties + schemaRegistry apply during route selection. It is type-agnostic: the wire schema-id resolves the concrete SpecificRecord with no caller-supplied type.
• matchValue reads the discriminator field from the deserialized GenericRecord using record.get(property). A non-Avro value or a missing field returns null (unmatched route, not an error).
• Serialization failures (missing schema, registry unreachable) surface as DeserializationException / failed Future and flow through the configured ErrorStrategy — never as an uncaught throw.

---

AvroModule

Dagger @Module that contributes ApicurioAvroSerdeProvider into the Set<KafkaSerdeProvider> multibinding.

@Module
public abstract class AvroModule {

    @Provides
    @Singleton
    @IntoSet
    static KafkaSerdeProvider avroSerdeProvider(KafkaConfig kafkaConfig) {
        return new ApicurioAvroSerdeProvider(kafkaConfig);
    }
}

Include alongside KafkaModule in the application @Component:

@Singleton
@Component(modules = {
    VertxModule.class,
    ConfigModule.class,
    KafkaModule.class,
    AvroModule.class,   // enables "avro" format
    AppModule.class,
    ServiceModule.class
})
interface AppComponent { ... }

---

Extension Points

This module is itself an extension of the KafkaSerdeProvider SPI. There are no additional extension points declared here.

---

Configuration

Minimal configuration

{
  "kafka": {
    "bootstrap.servers": "localhost:9092",
    "schemaRegistry": {
      "url": "http://localhost:8080/apis/registry/v3"
    }
  }
}

Per-consumer Avro format with custom serde properties

{
  "kafka": {
    "bootstrap.servers": "localhost:9092",
    "schemaRegistry": {
      "url": "http://schema-registry:8080/apis/registry/v3"
    },
    "consumers": {
      "order-created": {
        "topic": "order.created",
        "groupId": "order-processing",
        "format": "avro",
        "serdeProperties": {
          "apicurio.registry.auto-register": false
        }
      }
    }
  }
}

Per-producer-method Avro format

{
  "kafka": {
    "bootstrap.servers": "localhost:9092",
    "schemaRegistry": {
      "url": "http://schema-registry:8080/apis/registry/v3"
    },
    "producers": {
      "OrderEvents": {
        "methods": {
          "orderCreated": {
            "format": "avro"
          }
        }
      }
    }
  }
}

Config key reference

Key	Scope	Description
kafka.format	Global	Default format for all producers and consumers ("avro", "json")
kafka.schemaRegistry.url	Global	Schema Registry base URL
kafka.consumers.{name}.format	Per-consumer	Overrides global format for this consumer
kafka.consumers.{name}.serdeProperties.*	Per-consumer	Apicurio config overlaid on top of kafka.schemaRegistry.*
kafka.producers.{name}.format	Per-producer	Overrides global format for all methods of this producer
kafka.producers.{name}.serdeProperties.*	Per-producer	Apicurio config overlaid on top of kafka.schemaRegistry.*
kafka.producers.{name}.methods.{method}.format	Per-method	Overrides per-producer format for this producer method
kafka.producers.{name}.methods.{method}.serdeProperties.*	Per-method	Apicurio config for this method; merged last (highest priority)
serdeProperties."vertique.avro.trusted-packages"	Per-consumer / per-producer / per-method	Array of package names, or one comma-separated string. Each entry must have at least two segments (com.acme.events) and matches that package and its subpackages. Wildcards (*, com.acme.*), single-segment roots and JDK/library roots (java, javax, jakarta, jdk, sun, com.sun, org.apache, org.springframework) are rejected at startup. List only the dedicated package of your generated records: a prefix trusts every class under it

Format precedence (first match wins):

1. Per-method (or per-consumer) format key
2. Global kafka.format
3. Auto-detect: ApicurioAvroSerdeProvider.autoDetects(valueType) — true for SpecificRecord subtypes
4. "json" (built-in fallback)

---

Module Dagger Bindings

Binding	Kind	Description
KafkaSerdeProvider	@Provides @Singleton @IntoSet	ApicurioAvroSerdeProvider for format id "avro"

---

Verification

./mvnw -ntp -pl :vertique-kafka-avro -am test

---

Usage Example

Avro producer

@KafkaProducer
public interface OrderEvents {

    @Topic("order.created")
    Future<RecordMetadata> orderCreated(OrderCreated event);   // OrderCreated extends SpecificRecord
}

With AvroModule on the classpath and OrderCreated being a SpecificRecord, the "avro" format is auto-detected and no explicit format config is required. The serializer is built once at startup via KafkaProducerFactory; each orderCreated(...) call offloads serialization to a worker thread.

Avro consumer (auto-detected)

public class OrderServiceImpl implements OrderService {

    @KafkaSource(topic = "order.created", groupId = "order-processing")
    @Override
    public Future<Void> processOrder(OrderCreated event) {
        // event is deserialized from Avro; consumer verticle runs on worker thread
        return Future.succeededFuture();
    }
}

Avro routing interface (Model 3, matchProperty)

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

    @KafkaHandler(matchProperty = "eventType", matchValue = "order.created")
    @DispatchTo(service = OrderService.class, operation = "processOrder")
    void onOrderCreated(OrderCreated event);

    @KafkaHandler(matchProperty = "eventType", matchValue = "order.cancelled")
    @DispatchTo(service = OrderService.class, operation = "cancelOrder")
    void onOrderCancelled(OrderCancelled event);

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

For Model-3 matchProperty routing with Avro, the dispatcher uses the per-entry routingDeserializer once (type-agnostic, schema-id-driven), reads the discriminator field via matchValue, then reuses the record as the payload for the matched route.

---

Dependencies

Artifact	Scope	Purpose
dev.vertique:vertique-kafka-core	compile	KafkaSerdeProvider SPI, KafkaSerializer/KafkaDeserializer, DeserializationException
io.apicurio:apicurio-registry-avro-serde-kafka	compile	AvroKafkaSerializer/AvroKafkaDeserializer; Confluent-wire framing via Default4ByteIdHandler
org.apache.avro:avro	compile (transitive)	SpecificRecord, GenericRecord