Class AvroSerde<T>

java.lang.Object
io.krabka.streams.schema.AvroSerde<T>
Type Parameters:
T - the Avro record type carried by the serde
All Implemented Interfaces:
Closeable, AutoCloseable, org.apache.kafka.common.serialization.Serde<T>
public final class AvroSerde<T> extends Object
A Kafka serde for Confluent-framed Avro values.

Three data models are supported: generated SpecificRecord classes (forValue(Class, SchemaCache) and forKey(Class, SchemaCache)), schema-driven GenericRecord values (generic(Schema, SchemaCache, Role)), and ordinary Java classes through Avro reflection (reflect(Class, SchemaCache, Role)). All of them frame records with the Confluent wire format and resolve schema IDs through a shared SchemaCache.

Deserialization performs full Avro schema resolution: the record is read with the writer schema registered under the frame's schema ID and projected onto this serde's reader schema, so old and new record versions interoperate within Avro's compatibility rules.

Example

var client = new KrabkaSchemaRegistryClient(URI.create("http://localhost:8081"));
var cache = new SchemaCache(client);

var serde = AvroSerde.forValue(Order.class, cache);
serde.registerSubject("orders");
cache.prewarm().join();

byte[] bytes = serde.serializer().serialize("orders", order);
Order roundTripped = serde.deserializer().deserialize("orders", bytes);

  • Method Summary

    Modifier and Type
    Method
    Description
    protected final SchemaCache
    cache()
    Returns the schema cache this serde resolves subjects and writer schemas with.
    protected T
    deserializeBody(int schemaId, byte[] body)
    Decodes one body into a value using the writer schema registered under the ID.
    final org.apache.kafka.common.serialization.Deserializer<T>
    deserializer()
    Returns the deserializer half of this serde.
    static <T extends org.apache.avro.specific.SpecificRecord>
    AvroSerde<T>
    forKey(Class<T> type, SchemaCache cache)
    Creates a key serde for a generated Avro class.
    static <T extends org.apache.avro.specific.SpecificRecord>
    AvroSerde<T>
    forKey(Class<T> type, SchemaCache cache, SubjectNameStrategy strategy)
    Creates a key serde for a generated Avro class with a custom subject strategy.
    static <T extends org.apache.avro.specific.SpecificRecord>
    AvroSerde<T>
    forValue(Class<T> type, SchemaCache cache)
    Creates a value serde for a generated Avro class.
    static <T extends org.apache.avro.specific.SpecificRecord>
    AvroSerde<T>
    forValue(Class<T> type, SchemaCache cache, SubjectNameStrategy strategy)
    Creates a value serde for a generated Avro class with a custom subject strategy.
    protected byte[]
    frame(int schemaId, byte[] body)
    Wraps body bytes in the Confluent wire format.
    static AvroSerde<org.apache.avro.generic.GenericRecord>
    generic(org.apache.avro.Schema schema, SchemaCache cache, Role role)
    Creates a serde for schema-driven GenericRecord values.
    static AvroSerde<org.apache.avro.generic.GenericRecord>
    generic(org.apache.avro.Schema schema, SchemaCache cache, Role role, SubjectNameStrategy strategy)
    Creates a GenericRecord serde with a custom subject strategy.
    static <T> AvroSerde<T>
    reflect(Class<T> type, SchemaCache cache, Role role)
    Creates a serde backed by Avro reflection for ordinary Java classes.
    static <T> AvroSerde<T>
    reflect(Class<T> type, SchemaCache cache, Role role, SubjectNameStrategy strategy)
    Creates a reflection-based serde with a custom subject strategy.
    final void
    registerSubject(String topic)
    Adds this serde's subject to the cache prewarm set.
    protected byte[]
    serializeBody(T value)
    Encodes one value into the schema-format-specific body bytes, without framing.
    final org.apache.kafka.common.serialization.Serializer<T>
    serializer()
    Returns the serializer half of this serde.
    protected ConfluentWireFormat.Frame
    unframe(byte[] bytes)
    Splits framed record bytes into the schema ID and body.

    Methods inherited from class Object

    clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

    Methods inherited from interface org.apache.kafka.common.serialization.Serde

    close, configure

  • Method Details

    • forValue

      public static <T extends org.apache.avro.specific.SpecificRecord> AvroSerde<T> forValue(Class<T> type, SchemaCache cache)
      Creates a value serde for a generated Avro class.
      Type Parameters:
      T - the generated Avro record type
      Parameters:
      type - the generated class, whose embedded schema becomes the reader schema
      cache - the cache that resolves subjects and writer schemas
      Returns:
      a value serde for type

    • forValue

      public static <T extends org.apache.avro.specific.SpecificRecord> AvroSerde<T> forValue(Class<T> type, SchemaCache cache, SubjectNameStrategy strategy)
      Creates a value serde for a generated Avro class with a custom subject strategy.
      Type Parameters:
      T - the generated Avro record type
      Parameters:
      type - the generated class, whose embedded schema becomes the reader schema
      cache - the cache that resolves subjects and writer schemas
      strategy - the subject naming strategy that overrides the cache default
      Returns:
      a value serde for type

    • forKey

      public static <T extends org.apache.avro.specific.SpecificRecord> AvroSerde<T> forKey(Class<T> type, SchemaCache cache)
      Creates a key serde for a generated Avro class.
      Type Parameters:
      T - the generated Avro record type
      Parameters:
      type - the generated class, whose embedded schema becomes the reader schema
      cache - the cache that resolves subjects and writer schemas
      Returns:
      a key serde for type

    • forKey

      public static <T extends org.apache.avro.specific.SpecificRecord> AvroSerde<T> forKey(Class<T> type, SchemaCache cache, SubjectNameStrategy strategy)
      Creates a key serde for a generated Avro class with a custom subject strategy.
      Type Parameters:
      T - the generated Avro record type
      Parameters:
      type - the generated class, whose embedded schema becomes the reader schema
      cache - the cache that resolves subjects and writer schemas
      strategy - the subject naming strategy that overrides the cache default
      Returns:
      a key serde for type

    • generic

      public static AvroSerde<org.apache.avro.generic.GenericRecord> generic(org.apache.avro.Schema schema, SchemaCache cache, Role role)
      Creates a serde for schema-driven GenericRecord values.
      Parameters:
      schema - the reader schema for every record
      cache - the cache that resolves subjects and writer schemas
      role - whether the serde handles record keys or values
      Returns:
      a generic-record serde for schema

    • generic

      public static AvroSerde<org.apache.avro.generic.GenericRecord> generic(org.apache.avro.Schema schema, SchemaCache cache, Role role, SubjectNameStrategy strategy)
      Creates a GenericRecord serde with a custom subject strategy.
      Parameters:
      schema - the reader schema for every record
      cache - the cache that resolves subjects and writer schemas
      role - whether the serde handles record keys or values
      strategy - the subject naming strategy that overrides the cache default
      Returns:
      a generic-record serde for schema

    • reflect

      public static <T> AvroSerde<T> reflect(Class<T> type, SchemaCache cache, Role role)
      Creates a serde backed by Avro reflection for ordinary Java classes.

      The reader schema is derived from the class with ReflectData, so the class needs no generated code and no Avro annotations for simple shapes.

      Type Parameters:
      T - the reflected Java type
      Parameters:
      type - the Java class to derive the schema from
      cache - the cache that resolves subjects and writer schemas
      role - whether the serde handles record keys or values
      Returns:
      a reflection-based serde for type

    • reflect

      public static <T> AvroSerde<T> reflect(Class<T> type, SchemaCache cache, Role role, SubjectNameStrategy strategy)
      Creates a reflection-based serde with a custom subject strategy.
      Type Parameters:
      T - the reflected Java type
      Parameters:
      type - the Java class to derive the schema from
      cache - the cache that resolves subjects and writer schemas
      role - whether the serde handles record keys or values
      strategy - the subject naming strategy that overrides the cache default
      Returns:
      a reflection-based serde for type

    • serializeBody

      protected byte[] serializeBody(T value) throws Exception
      Encodes one value into the schema-format-specific body bytes, without framing.
      Parameters:
      value - the non-null value to encode
      Returns:
      the encoded body
      Throws:
      Exception - if the value cannot be encoded

    • deserializeBody

      protected T deserializeBody(int schemaId, byte[] body) throws Exception
      Decodes one body into a value using the writer schema registered under the ID.
      Parameters:
      schemaId - the schema ID taken from the frame header
      body - the body bytes that followed the frame header
      Returns:
      the decoded value
      Throws:
      Exception - if the body cannot be decoded

    • registerSubject

      public final void registerSubject(String topic)
      Adds this serde's subject to the cache prewarm set.

      Call this once per topic the serde will read or write, then resolve every registered subject with SchemaCache.prewarm() before the first record is serialized. Registration is idempotent by subject.

      Parameters:
      topic - the Kafka topic whose subject should be resolved during prewarm

    • cache

      protected final SchemaCache cache()
      Returns the schema cache this serde resolves subjects and writer schemas with.
      Returns:
      the schema cache supplied at construction

    • serializer

      public final org.apache.kafka.common.serialization.Serializer<T> serializer()
      Returns the serializer half of this serde.

      The serializer frames each non-null value with the Confluent wire format using the schema ID resolved during prewarm. It throws SerializationException when the subject has not been resolved yet or when the body cannot be encoded. A null value serializes to null.

      Specified by:
      serializer in interface org.apache.kafka.common.serialization.Serde<T>
      Returns:
      the reusable, thread-safe serializer

    • deserializer

      public final org.apache.kafka.common.serialization.Deserializer<T> deserializer()
      Returns the deserializer half of this serde.

      The deserializer unframes the Confluent header, resolves the writer schema for the frame's schema ID through the cache, and decodes the body. It throws SchemaFetchPendingException while an unknown writer schema is being fetched and SerializationException for malformed input. Null input deserializes to null.

      Specified by:
      deserializer in interface org.apache.kafka.common.serialization.Serde<T>
      Returns:
      the reusable, thread-safe deserializer

    • frame

      protected byte[] frame(int schemaId, byte[] body)
      Wraps body bytes in the Confluent wire format.
      Parameters:
      schemaId - the resolved schema ID for the subject
      body - the encoded body
      Returns:
      the framed record bytes

    • unframe

      protected ConfluentWireFormat.Frame unframe(byte[] bytes)
      Splits framed record bytes into the schema ID and body.
      Parameters:
      bytes - the framed record bytes
      Returns:
      the decoded frame