Class JsonSchemaSerde<T>

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

Values are bound to and from JSON with Jackson, so any Jackson-compatible class works, including Java records. Optional validation checks the serialized document against the local schema when writing and against the registered writer schema when reading; a validation failure throws SerializationException.

The schema dialect is detected from the document's $schema keyword and falls back to draft 2020-12; pass an explicit SpecificationVersion through the long factory overloads to pin it.

Example

record Order(String user, long amount) {}

String orderSchema = """
    {"type": "object",
     "properties": {"user": {"type": "string"}, "amount": {"type": "integer"}},
     "required": ["user", "amount"]}""";

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

byte[] bytes = serde.serializer().serialize("orders", new Order("ada", 7));

  • 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> JsonSchemaSerde<T>
    forKey(Class<T> type, String schema, SchemaCache cache, boolean validate)
    Creates a key serde with a default Jackson mapper.
    static <T> JsonSchemaSerde<T>
    forKey(Class<T> type, String schema, SchemaCache cache, boolean validate, com.networknt.schema.SpecificationVersion dialect, SubjectNameStrategy strategy, com.fasterxml.jackson.databind.ObjectMapper objectMapper)
    Creates a fully configured key serde.
    static <T> JsonSchemaSerde<T>
    forValue(Class<T> type, String schema, SchemaCache cache, boolean validate)
    Creates a value serde with a default Jackson mapper.
    static <T> JsonSchemaSerde<T>
    forValue(Class<T> type, String schema, SchemaCache cache, boolean validate, com.fasterxml.jackson.databind.ObjectMapper objectMapper)
    Creates a value serde with a caller-supplied Jackson mapper.
    static <T> JsonSchemaSerde<T>
    forValue(Class<T> type, String schema, SchemaCache cache, boolean validate, com.networknt.schema.SpecificationVersion dialect, SubjectNameStrategy strategy, com.fasterxml.jackson.databind.ObjectMapper objectMapper)
    Creates a fully configured value serde.
    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> JsonSchemaSerde<T> forValue(Class<T> type, String schema, SchemaCache cache, boolean validate)
      Creates a value serde with a default Jackson mapper.
      Type Parameters:
      T - the Jackson-compatible value type
      Parameters:
      type - the class values are bound to and from
      schema - the JSON Schema document registered for the subject
      cache - the cache that resolves subjects and writer schemas
      validate - whether to validate documents on both serialize and deserialize
      Returns:
      a value serde for type
      Throws:
      IllegalArgumentException - if schema is not valid JSON

    • forKey

      public static <T> JsonSchemaSerde<T> forKey(Class<T> type, String schema, SchemaCache cache, boolean validate)
      Creates a key serde with a default Jackson mapper.
      Type Parameters:
      T - the Jackson-compatible key type
      Parameters:
      type - the class keys are bound to and from
      schema - the JSON Schema document registered for the subject
      cache - the cache that resolves subjects and writer schemas
      validate - whether to validate documents on both serialize and deserialize
      Returns:
      a key serde for type
      Throws:
      IllegalArgumentException - if schema is not valid JSON

    • forValue

      public static <T> JsonSchemaSerde<T> forValue(Class<T> type, String schema, SchemaCache cache, boolean validate, com.fasterxml.jackson.databind.ObjectMapper objectMapper)
      Creates a value serde with a caller-supplied Jackson mapper.
      Type Parameters:
      T - the Jackson-compatible value type
      Parameters:
      type - the class values are bound to and from
      schema - the JSON Schema document registered for the subject
      cache - the cache that resolves subjects and writer schemas
      validate - whether to validate documents on both serialize and deserialize
      objectMapper - the mapper used for binding and validation input
      Returns:
      a value serde for type
      Throws:
      IllegalArgumentException - if schema is not valid JSON

    • forValue

      public static <T> JsonSchemaSerde<T> forValue(Class<T> type, String schema, SchemaCache cache, boolean validate, com.networknt.schema.SpecificationVersion dialect, SubjectNameStrategy strategy, com.fasterxml.jackson.databind.ObjectMapper objectMapper)
      Creates a fully configured value serde.
      Type Parameters:
      T - the Jackson-compatible value type
      Parameters:
      type - the class values are bound to and from
      schema - the JSON Schema document registered for the subject
      cache - the cache that resolves subjects and writer schemas
      validate - whether to validate documents on both serialize and deserialize
      dialect - the JSON Schema dialect, or null to detect it from the document
      strategy - the subject naming strategy, or null for the cache default
      objectMapper - the mapper used for binding and validation input
      Returns:
      a value serde for type
      Throws:
      IllegalArgumentException - if schema is not valid JSON

    • forKey

      public static <T> JsonSchemaSerde<T> forKey(Class<T> type, String schema, SchemaCache cache, boolean validate, com.networknt.schema.SpecificationVersion dialect, SubjectNameStrategy strategy, com.fasterxml.jackson.databind.ObjectMapper objectMapper)
      Creates a fully configured key serde.
      Type Parameters:
      T - the Jackson-compatible key type
      Parameters:
      type - the class keys are bound to and from
      schema - the JSON Schema document registered for the subject
      cache - the cache that resolves subjects and writer schemas
      validate - whether to validate documents on both serialize and deserialize
      dialect - the JSON Schema dialect, or null to detect it from the document
      strategy - the subject naming strategy, or null for the cache default
      objectMapper - the mapper used for binding and validation input
      Returns:
      a key serde for type
      Throws:
      IllegalArgumentException - if schema is not valid JSON

    • 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