Class AvroBatchCodec

java.lang.Object
io.krabka.streams.columnar.schema.AvroBatchCodec
All Implemented Interfaces:
BatchCodec
public final class AvroBatchCodec extends Object implements BatchCodec
Decodes registry-framed Avro topics into Arrow batches and encodes them back.

The codec composes an AvroSerde with an AvroRowBridge: each record value is unframed, resolved from its writer schema onto the fixed reader schema, and written as one row whose columns follow the reader schema — structs, lists, maps, decimals, and timestamps as native Arrow types. The reserved __key, __timestamp, __partition, __offset, and __headers columns are appended as for every RowCodec. Because the Arrow schema derives from the reader schema alone, a new writer schema mid-stream never changes the columns.

Subjects resolve through the shared SchemaCache: call registerSubject(String) for every topic the codec reads or writes, then prewarm the cache before processing. An unknown writer schema ID during decode starts one background fetch and throws the cache's retriable SchemaFetchPendingException, exactly as ordinary consumers experience.

Example

var codec = AvroBatchCodec.generic(orderSchema, cache, allocator);
codec.registerSubject("orders");
cache.prewarm().join();

var topology = new ColumnarTopology(allocator);
var source = topology.addSource("orders", List.of("orders"), codec);

  • Method Summary

    Modifier and Type
    Method
    Description
    org.apache.arrow.vector.types.pojo.Schema
    arrowSchema()
    Returns the payload Arrow schema of decoded batches, without the reserved metadata columns.
    org.apache.arrow.vector.VectorSchemaRoot
    decode(String topic, List<ConsumedRecord> records)
    Decodes records, resolving writer schemas through the topic's subject.
    org.apache.arrow.vector.VectorSchemaRoot
    decode(List<ConsumedRecord> records)
    Decodes records with an empty topic name.
    List<ProduceRecord>
    encode(String topic, org.apache.arrow.vector.VectorSchemaRoot batch)
    Encodes a batch into one registry-framed record per row.
    List<ProduceRecord>
    encode(org.apache.arrow.vector.VectorSchemaRoot batch)
    Encodes a batch with an empty topic name.
    static <T extends org.apache.avro.specific.SpecificRecord>
    AvroBatchCodec
    forValue(Class<T> type, SchemaCache cache, org.apache.arrow.memory.BufferAllocator allocator)
    Creates a codec for a generated Avro class.
    static AvroBatchCodec
    generic(org.apache.avro.Schema schema, SchemaCache cache, org.apache.arrow.memory.BufferAllocator allocator)
    Creates a codec for schema-driven GenericRecord values.
    void
    registerSubject(String topic)
    Interns the reader schema for a topic's value subject in the shared cache.

    Methods inherited from class Object

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

  • Method Details

    • generic

      public static AvroBatchCodec generic(org.apache.avro.Schema schema, SchemaCache cache, org.apache.arrow.memory.BufferAllocator allocator)
      Creates a codec for schema-driven GenericRecord values.
      Parameters:
      schema - the reader schema for every record; must be a record schema
      cache - the cache that resolves subjects and writer schemas
      allocator - the allocator that owns decoded batches
      Returns:
      a codec whose columns follow schema

    • forValue

      public static <T extends org.apache.avro.specific.SpecificRecord> AvroBatchCodec forValue(Class<T> type, SchemaCache cache, org.apache.arrow.memory.BufferAllocator allocator)
      Creates a codec for a generated Avro class.
      Type Parameters:
      T - the generated Avro record type
      Parameters:
      type - the generated class, whose embedded schema drives the columns
      cache - the cache that resolves subjects and writer schemas
      allocator - the allocator that owns decoded batches
      Returns:
      a codec whose columns follow the embedded schema of type

    • registerSubject

      public void registerSubject(String topic)
      Interns the reader schema for a topic's value subject in the shared cache.

      Call this once per topic before prewarming, exactly as with the underlying serde.

      Parameters:
      topic - the topic whose subject resolves the reader schema

    • arrowSchema

      public org.apache.arrow.vector.types.pojo.Schema arrowSchema()
      Returns the payload Arrow schema of decoded batches, without the reserved metadata columns.
      Returns:
      the Arrow schema derived from the reader schema

    • decode

      public org.apache.arrow.vector.VectorSchemaRoot decode(List<ConsumedRecord> records)
      Decodes records with an empty topic name.
      Specified by:
      decode in interface BatchCodec
      Parameters:
      records - the records of one topic partition batch
      Returns:
      the decoded batch; the caller must close it

    • decode

      public org.apache.arrow.vector.VectorSchemaRoot decode(String topic, List<ConsumedRecord> records)
      Decodes records, resolving writer schemas through the topic's subject.
      Specified by:
      decode in interface BatchCodec
      Parameters:
      topic - the topic the records were fetched from
      records - the records of one topic partition batch
      Returns:
      the decoded batch, one row per record, with metadata columns appended; the caller must close it

    • encode

      public List<ProduceRecord> encode(org.apache.arrow.vector.VectorSchemaRoot batch)
      Encodes a batch with an empty topic name.
      Specified by:
      encode in interface BatchCodec
      Parameters:
      batch - the batch to encode; the codec reads it and leaves it open
      Returns:
      the encoded records in row order

    • encode

      public List<ProduceRecord> encode(String topic, org.apache.arrow.vector.VectorSchemaRoot batch)
      Encodes a batch into one registry-framed record per row.
      Specified by:
      encode in interface BatchCodec
      Parameters:
      topic - the topic the records will be produced to
      batch - the batch to encode; the codec reads it and leaves it open
      Returns:
      the encoded records in row order