Class ProtobufSerde<T extends com.google.protobuf.Message>

java.lang.Object
io.krabka.streams.schema.ProtobufSerde<T>
Type Parameters:
T - the generated Protobuf message type
All Implemented Interfaces:
Closeable, AutoCloseable, org.apache.kafka.common.serialization.Serde<T>
public final class ProtobufSerde<T extends com.google.protobuf.Message> extends Object
A Kafka serde for Confluent-framed Protobuf messages.

The serde is created from a generated message's default instance. The .proto schema text registered with the registry is reconstructed from the message's file descriptor, and the Confluent Protobuf frame — including the message-index list that identifies nested messages — is written and read automatically.

On deserialization the writer's registered messageType is compared with the local message's full name; a mismatch throws SerializationException instead of silently parsing bytes into the wrong message.

Example

var serde = ProtobufSerde.forValue(Order.getDefaultInstance(), 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 com.google.protobuf.Message>
    ProtobufSerde<T>
    forKey(T defaultInstance, SchemaCache cache)
    Creates a key serde for a generated Protobuf message.
    static <T extends com.google.protobuf.Message>
    ProtobufSerde<T>
    forKey(T defaultInstance, SchemaCache cache, SubjectNameStrategy strategy)
    Creates a key serde with a custom subject strategy.
    static <T extends com.google.protobuf.Message>
    ProtobufSerde<T>
    forValue(T defaultInstance, SchemaCache cache)
    Creates a value serde for a generated Protobuf message.
    static <T extends com.google.protobuf.Message>
    ProtobufSerde<T>
    forValue(T defaultInstance, SchemaCache cache, SubjectNameStrategy strategy)
    Creates a value serde with a custom subject strategy.
    protected byte[]
    frame(int schemaId, byte[] body)
    Wraps body bytes in the Confluent wire format.
    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 com.google.protobuf.Message> ProtobufSerde<T> forValue(T defaultInstance, SchemaCache cache)
      Creates a value serde for a generated Protobuf message.
      Type Parameters:
      T - the generated Protobuf message type
      Parameters:
      defaultInstance - the message's default instance, for example Order.getDefaultInstance()
      cache - the cache that resolves subjects and writer schemas
      Returns:
      a value serde for the message type

    • forValue

      public static <T extends com.google.protobuf.Message> ProtobufSerde<T> forValue(T defaultInstance, SchemaCache cache, SubjectNameStrategy strategy)
      Creates a value serde with a custom subject strategy.
      Type Parameters:
      T - the generated Protobuf message type
      Parameters:
      defaultInstance - the message's default instance, for example Order.getDefaultInstance()
      cache - the cache that resolves subjects and writer schemas
      strategy - the subject naming strategy that overrides the cache default
      Returns:
      a value serde for the message type

    • forKey

      public static <T extends com.google.protobuf.Message> ProtobufSerde<T> forKey(T defaultInstance, SchemaCache cache)
      Creates a key serde for a generated Protobuf message.
      Type Parameters:
      T - the generated Protobuf message type
      Parameters:
      defaultInstance - the message's default instance, for example OrderKey.getDefaultInstance()
      cache - the cache that resolves subjects and writer schemas
      Returns:
      a key serde for the message type

    • forKey

      public static <T extends com.google.protobuf.Message> ProtobufSerde<T> forKey(T defaultInstance, SchemaCache cache, SubjectNameStrategy strategy)
      Creates a key serde with a custom subject strategy.
      Type Parameters:
      T - the generated Protobuf message type
      Parameters:
      defaultInstance - the message's default instance, for example OrderKey.getDefaultInstance()
      cache - the cache that resolves subjects and writer schemas
      strategy - the subject naming strategy that overrides the cache default
      Returns:
      a key serde for the message type

    • serializeBody

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

    • 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

    • 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

    • 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