Class ProtobufBatchCodec

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

The codec composes a ProtobufSerde with a ProtobufRowBridge: each record value is unframed and parsed with the message type's descriptor, then written as one row whose columns follow the descriptor — nested messages, repeated fields, maps, and google.protobuf.Timestamp as native Arrow types. The reserved __key, __timestamp, __partition, __offset, and __headers columns are appended as for every RowCodec. The Arrow schema derives from the local descriptor alone, so a new writer schema mid-stream never changes the columns; unknown fields are dropped by the descriptor-driven rebuild.

Subjects resolve through the shared SchemaCache: call registerSubject(String) for every topic the codec reads or writes, then prewarm the cache before processing.

Example

var codec = ProtobufBatchCodec.of(Order.getDefaultInstance(), 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 the subject through the topic name.
    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 com.google.protobuf.Message>
    ProtobufBatchCodec
    of(T defaultInstance, SchemaCache cache, org.apache.arrow.memory.BufferAllocator allocator)
    Creates a codec for one message type.
    void
    registerSubject(String topic)
    Interns the message 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

    • of

      public static <T extends com.google.protobuf.Message> ProtobufBatchCodec of(T defaultInstance, SchemaCache cache, org.apache.arrow.memory.BufferAllocator allocator)
      Creates a codec for one message type.
      Type Parameters:
      T - the message type
      Parameters:
      defaultInstance - the default instance of the message type, for example Order.getDefaultInstance()
      cache - the cache that resolves subjects and writer schemas
      allocator - the allocator that owns decoded batches
      Returns:
      a codec whose columns follow the message descriptor

    • registerSubject

      public void registerSubject(String topic)
      Interns the message 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 message 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 message descriptor

    • 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 the subject through the topic name.
      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