package schema

import "github.com/krabka-io/krabka-streams-go/schema"

Package schema talks to a Confluent-compatible schema registry and provides Avro, Protobuf, and JSON Schema serdes over the Confluent wire format.

The package splits registry I/O from serialization: RegistryClient performs HTTP calls, SchemaCache holds resolved schema IDs and writer schemas in memory, and the serdes only ever read from the cache. Resolve everything you can before processing starts with SchemaCache.Prewarm; a consumer meeting an unknown writer schema ID mid-stream triggers a single background fetch and a retriable FetchPendingError.

Index

Constants

const Magic byte = 0

Magic is the first byte of every Confluent-framed record.

Functions

func Encode source

func Encode(schemaID int, body []byte) []byte

Encode frames body bytes with the magic byte and schema ID.

Every framed record starts with a five-byte header: the Magic byte followed by the schema ID as a big-endian 32-bit integer.

Example 
package main

import (
    "fmt"

    "github.com/krabka-io/krabka-streams-go/schema"
)

func main() {
    frame := schema.Encode(258, []byte("body"))

    decoded, err := schema.Decode(frame)
    if err != nil {
        panic(err)
    }
    fmt.Println(decoded.SchemaID, string(decoded.Body))
}

Output:

258 body

func EncodeProtobuf source

func EncodeProtobuf(schemaID int, messageIndexes []int, body []byte) ([]byte, error)

EncodeProtobuf frames Protobuf body bytes, including the message-index list encoded as zigzag varints between the header and the body.

The indexes identify the message within its .proto file: each entry is the message's position at its nesting level, outermost first. The common [0] path (first top-level message) is written in its compact single-byte form. The index path must not be empty.

func PrintProtoFile source

func PrintProtoFile(file protoreflect.FileDescriptor) string

PrintProtoFile reconstructs .proto source text from a file descriptor.

The output covers the syntax declaration, package, imports, options, nested definitions, enums, oneofs, services, extensions, maps, and proto2 groups. It is the schema text the Protobuf serde registers with the registry.

func TopicNameStrategy source

func TopicNameStrategy(topic string, role Role) string

TopicNameStrategy returns the Confluent topic naming rule: a key schema maps to "<topic>-key" and a value schema to "<topic>-value". This matches Confluent's default TopicNameStrategy, so applications sharing topics with Confluent-configured clients resolve the same subjects.

Example 
package main

import (
    "fmt"

    "github.com/krabka-io/krabka-streams-go/schema"
)

func main() {
    fmt.Println(schema.TopicNameStrategy("orders", schema.RoleKey))
    fmt.Println(schema.TopicNameStrategy("orders", schema.RoleValue))
}

Output:

orders-key
orders-value

Types

type AvroSerde source

type AvroSerde[T any] struct {
    // contains filtered or unexported fields
}

AvroSerde is a Serde that encodes values as Confluent-framed Avro binary.

The schema registered with the registry is the Avro canonical parsing form, which strips documentation, ordering, and aliases. Two schemas that differ only in those respects therefore resolve to the same registry ID.

Writing uses binary encoding against the serde's own reader schema. Reading resolves the writer schema from the cache onto the reader schema, so full Avro schema resolution applies: added fields with defaults, dropped fields, promoted numeric types, and renamed records with aliases all work as Avro specifies.

func NewAvroSerde source

func NewAvroSerde[T any](schemaText string, cache *SchemaCache, role Role, options ...SerdeOption) (*AvroSerde[T], error)

NewAvroSerde creates a serde that maps Avro data onto the Go type T using hamba/avro struct tags, the Go equivalent of Avro's reflect and specific data models. The schema text must parse as an Avro schema.

func NewGenericAvroSerde source

func NewGenericAvroSerde(schemaText string, cache *SchemaCache, role Role, options ...SerdeOption) (*AvroSerde[any], error)

NewGenericAvroSerde creates a serde over Avro's generic representation: records become map[string]any, and scalars their natural Go types.

func (*AvroSerde) Deserialize source

func (s *AvroSerde) Deserialize(topic string, data []byte) (T, error)

func (*AvroSerde[T]) ReaderSchema source

func (s *AvroSerde[T]) ReaderSchema() avro.Schema

ReaderSchema returns the serde's parsed reader schema.

func (*AvroSerde) RegisterSubject source

func (s *AvroSerde) RegisterSubject(topic string)

func (*AvroSerde) Serialize source

func (s *AvroSerde) Serialize(topic string, value T) ([]byte, error)

type CompatibilityMode source

type CompatibilityMode int

CompatibilityMode selects the direction of a local compatibility check. 

const (
    // Backward requires that the candidate schema can read data written with
    // the previous schema.
    Backward CompatibilityMode = iota

    // Forward requires that the previous schema can read data written with
    // the candidate schema.
    Forward

    // Full requires both backward and forward compatibility.
    Full
)

func (CompatibilityMode) String source

func (m CompatibilityMode) String() string

String returns "BACKWARD", "FORWARD", or "FULL".

type CompatibilityResult source

type CompatibilityResult struct {
    // Compatible is true when no incompatibility was found.
    Compatible bool

    // Incompatibilities lists every found incompatibility, prefixed with the
    // direction that failed.
    Incompatibilities []string
}

CompatibilityResult reports the outcome of a local compatibility check.

func AvroCompatibility source

func AvroCompatibility(previousSchema, candidateSchema string, mode CompatibilityMode) (CompatibilityResult, error)

AvroCompatibility checks a previous and candidate Avro schema without contacting a registry, delegating to Avro reader/writer compatibility rules.

func JSONCompatibility source

func JSONCompatibility(previousSchema, candidateSchema string, mode CompatibilityMode) (CompatibilityResult, error)

JSONCompatibility checks a previous and candidate JSON Schema without contacting a registry. It checks type narrowing, required fields, and closed object properties.

func ProtobufCompatibility source

func ProtobufCompatibility(previous, candidate protoreflect.FileDescriptor, mode CompatibilityMode) CompatibilityResult

ProtobufCompatibility checks a previous and candidate Protobuf file descriptor without contacting a registry. It compares message fields by wire type, cardinality, and required-field presence.

type FetchPendingError source

type FetchPendingError struct {
    // SchemaID is the schema ID whose writer schema is being fetched.
    SchemaID int
}

FetchPendingError signals that a background writer-schema fetch has not completed.

SchemaCache.WriterSchema never blocks. When a deserializer meets a schema ID that is not cached yet, the cache starts one asynchronous registry fetch and returns this error. Treat it as retriable: by the time the record is retried, the fetch has usually completed and deserialization succeeds.

func (*FetchPendingError) Error source

func (e *FetchPendingError) Error() string

Error names the pending schema ID.

func (*FetchPendingError) Retriable source

func (e *FetchPendingError) Retriable() bool

Retriable marks the error as safe to retry.

type FetchedSchema source

type FetchedSchema struct {
    // Schema is the schema text.
    Schema string

    // MessageType is the Protobuf message full name, or "" for other formats.
    MessageType string

    // References are the schemas this schema refers to, by name and subject
    // version.
    References []SchemaReference
}

FetchedSchema is a schema fetched by ID, with unresolved references.

type Frame source

type Frame struct {
    // SchemaID is the registry schema ID from the header.
    SchemaID int

    // Body is the record body that followed the header.
    Body []byte
}

Frame is a decoded Confluent frame. The body is defensively copied at construction, so a frame never aliases a caller's slice.

func Decode source

func Decode(data []byte) (Frame, error)

Decode splits framed record bytes into the schema ID and body.

It fails if the input is shorter than five bytes or does not start with Magic.

type JSONSchemaSerde source

type JSONSchemaSerde[T any] struct {
    // contains filtered or unexported fields
}

JSONSchemaSerde is a Serde that encodes values as Confluent-framed JSON validated against a JSON Schema.

When validation is enabled, serialization checks the document against the serde's local schema and deserialization checks it against the fetched writer schema. Compiled validators are cached per schema ID.

The serde detects Draft 4, 6, 7, 2019-09, or 2020-12 from $schema; schemas without it default to 2020-12.

func NewJSONSchemaSerde source

func NewJSONSchemaSerde[T any](schemaText string, cache *SchemaCache, role Role, validate bool, options ...SerdeOption) (*JSONSchemaSerde[T], error)

NewJSONSchemaSerde creates a serde that maps documents onto T with encoding/json. The validate flag enables schema validation on both paths.

func (*JSONSchemaSerde) Deserialize source

func (s *JSONSchemaSerde) Deserialize(topic string, data []byte) (T, error)

func (*JSONSchemaSerde) RegisterSubject source

func (s *JSONSchemaSerde) RegisterSubject(topic string)

func (*JSONSchemaSerde) Serialize source

func (s *JSONSchemaSerde) Serialize(topic string, value T) ([]byte, error)

type Kind source

type Kind int

Kind enumerates the schema formats supported by the registry client.

The kind is sent as the Confluent schemaType field when registering or looking up a schema. Avro is the registry default and is therefore transmitted without an explicit type name. 

const (
    // KindAvro is an Apache Avro schema, the registry default format.
    KindAvro Kind = iota

    // KindProtobuf is a Protocol Buffers file descriptor rendered as .proto text.
    KindProtobuf

    // KindJSON is a JSON Schema document.
    KindJSON
)

type PrewarmReport source

type PrewarmReport struct {
    // Resolved maps each successful subject to its resolved schema ID.
    Resolved map[string]int

    // Failures maps each failed subject to its failure cause.
    Failures map[string]error
}

PrewarmReport is the per-subject outcome of SchemaCache.PrewarmReport.

func (PrewarmReport) Successful source

func (r PrewarmReport) Successful() bool

Successful reports whether every subject resolved.

type ProtobufFrame source

type ProtobufFrame struct {
    // SchemaID is the registry schema ID from the header.
    SchemaID int

    // MessageIndexes is the message-index path, outermost first.
    MessageIndexes []int

    // Body is the serialized Protobuf message that followed the indexes.
    Body []byte
}

ProtobufFrame is a decoded Confluent Protobuf frame. The body is defensively copied at construction, so a frame never aliases a caller's slice.

func DecodeProtobuf source

func DecodeProtobuf(data []byte) (ProtobufFrame, error)

DecodeProtobuf splits framed Protobuf record bytes into schema ID, message indexes, and body.

It fails if the input is shorter than five bytes, does not start with Magic, or carries a malformed message-index list.

type ProtobufSerde source

type ProtobufSerde[T proto.Message] struct {
    // contains filtered or unexported fields
}

ProtobufSerde is a Serde that encodes Protobuf messages with Confluent Protobuf framing, including the message-index path.

The serde derives three things from its prototype message: the schema text, printed from the message's file descriptor with PrintProtoFile; the messageType, which is the fully qualified message name; and the full message-index path from the top-level parent to a nested message.

Deserialization enforces a type check: if the cache holds a messageType for the writer's schema ID and it differs from the local message's full name, deserialization fails. That check is skipped when the registry supplied no messageType.

func NewProtobufSerde source

func NewProtobufSerde[T proto.Message](prototype T, cache *SchemaCache, role Role, options ...SerdeOption) *ProtobufSerde[T]

NewProtobufSerde creates a serde for the message type of prototype, which is typically the zero value of a generated message type, for example &ordersv1.Order{}.

func (*ProtobufSerde) Deserialize source

func (s *ProtobufSerde) Deserialize(topic string, data []byte) (T, error)

func (*ProtobufSerde) RegisterSubject source

func (s *ProtobufSerde) RegisterSubject(topic string)

func (*ProtobufSerde) Serialize source

func (s *ProtobufSerde) Serialize(topic string, value T) ([]byte, error)

type RegisterMode source

type RegisterMode int

RegisterMode defines how prewarming resolves a schema ID.

SchemaCache.Prewarm resolves every interned subject to a schema ID by talking to the registry once. The mode chooses the registry operation used for that resolution. 

const (
    // AutoRegister registers the local schema, creating a new subject version
    // when the schema is unknown to the registry. Registration is idempotent:
    // re-registering an existing schema returns its existing ID.
    AutoRegister RegisterMode = iota

    // LookupOnly looks the local schema up under the subject and fails
    // prewarming when the registry does not already know it. Use this in
    // environments where schemas are registered by a deployment pipeline
    // rather than by applications.
    LookupOnly

    // UseLatest uses the latest registered version of the subject regardless
    // of the local schema text. The writer schema recorded for the resolved ID
    // is still the local schema, so the local and latest schemas must be
    // compatible.
    UseLatest
)

type RegisteredSchema source

type RegisteredSchema struct {
    // ID is the schema's global registry ID.
    ID  int

    // Version is the version number within the subject.
    Version int

    // Schema is the schema text.
    Schema string

    // SchemaType is the registry schema type name, or "" for Avro.
    SchemaType string

    // MessageType is the Protobuf message full name, or "" for other formats.
    MessageType string

    // References are the schemas this schema refers to.
    References []SchemaReference
}

RegisteredSchema is one registered version of a subject.

type RegistryClient source

type RegistryClient struct {
    // contains filtered or unexported fields
}

RegistryClient is a client for the Confluent Schema Registry REST API.

Transport errors, HTTP 429, and 5xx responses are retried up to the configured retry count. Failures are reported as *RegistryError.

Applications rarely call this client directly for serialization; they hand it to a SchemaCache, which layers caching and prewarming on top. Direct use suits tooling: registering schemas from a pipeline, checking compatibility levels, or cleaning up subjects.

The client is safe for concurrent use.

func NewRegistryClient source

func NewRegistryClient(baseURL string, options ...RegistryClientOption) (*RegistryClient, error)

NewRegistryClient creates a client for the registry at baseURL, for example "http://localhost:8081". Trailing slashes are normalized and context paths are preserved.

func (*RegistryClient) Compatibility source

func (c *RegistryClient) Compatibility(ctx context.Context) (string, error)

Compatibility reads the registry's global compatibility level.

func (*RegistryClient) DeleteSubject source

func (c *RegistryClient) DeleteSubject(ctx context.Context, subject string, permanent bool) ([]int, error)

DeleteSubject deletes a subject and all of its versions, returning the deleted version numbers. When permanent is true the subject is hard-deleted instead of soft-deleted.

func (*RegistryClient) DeleteVersion source

func (c *RegistryClient) DeleteVersion(ctx context.Context, subject string, version int, permanent bool) (int, error)

DeleteVersion deletes one version of a subject, returning the deleted version number. When permanent is true the version is hard-deleted instead of soft-deleted.

func (*RegistryClient) Latest source

func (c *RegistryClient) Latest(ctx context.Context, subject string) (RegisteredSchema, error)

Latest fetches the latest registered version of a subject.

func (*RegistryClient) LatestID source

func (c *RegistryClient) LatestID(ctx context.Context, subject string) (int, error)

LatestID fetches the schema ID of a subject's latest version.

func (*RegistryClient) Lookup source

func (c *RegistryClient) Lookup(ctx context.Context, subject string, kind Kind, schemaText, messageType string, references ...SchemaReference) (int, error)

Lookup looks up the ID of an already registered schema, optionally with schema references. It fails with HTTP 404 when the schema is not registered under the subject.

func (*RegistryClient) Register source

func (c *RegistryClient) Register(ctx context.Context, subject string, kind Kind, schemaText, messageType string, references ...SchemaReference) (int, error)

Register registers a schema under a subject, optionally with schema references, and returns its ID.

Registration is idempotent: registering a schema the subject already has returns the existing ID. messageType is the Protobuf message full name and "" for other formats.

func (*RegistryClient) ResolvedSchemaByID source

func (c *RegistryClient) ResolvedSchemaByID(ctx context.Context, schemaID int) (ResolvedSchema, error)

ResolvedSchemaByID fetches a schema by ID and recursively resolves its references.

Each referenced subject version is fetched, and its own references are resolved transitively. Reference cycles are tolerated: a subject version already being resolved on the current path is skipped.

func (*RegistryClient) SchemaByID source

func (c *RegistryClient) SchemaByID(ctx context.Context, schemaID int) (FetchedSchema, error)

SchemaByID fetches a schema by its global ID. References are not resolved to their own schema texts; use RegistryClient.ResolvedSchemaByID for that.

func (*RegistryClient) SetCompatibility source

func (c *RegistryClient) SetCompatibility(ctx context.Context, level string) (string, error)

SetCompatibility sets the registry's global compatibility level and returns the level the registry acknowledged.

func (*RegistryClient) SetSubjectCompatibility source

func (c *RegistryClient) SetSubjectCompatibility(ctx context.Context, subject, level string) (string, error)

SetSubjectCompatibility sets a subject's compatibility level and returns the level the registry acknowledged.

func (*RegistryClient) SubjectCompatibility source

func (c *RegistryClient) SubjectCompatibility(ctx context.Context, subject string) (string, error)

SubjectCompatibility reads a subject's compatibility level, for example "BACKWARD".

func (*RegistryClient) Subjects source

func (c *RegistryClient) Subjects(ctx context.Context) ([]string, error)

Subjects lists all subject names.

func (*RegistryClient) Version source

func (c *RegistryClient) Version(ctx context.Context, subject string, version int) (RegisteredSchema, error)

Version fetches one specific version of a subject.

func (*RegistryClient) Versions source

func (c *RegistryClient) Versions(ctx context.Context, subject string) ([]int, error)

Versions lists a subject's version numbers in registry order.

type RegistryClientOption source

type RegistryClientOption func(*RegistryClient)

RegistryClientOption configures a RegistryClient.

func WithBasicAuth source

func WithBasicAuth(username, password string) RegistryClientOption

WithBasicAuth sends HTTP basic authentication with every request.

func WithHTTPClient source

func WithHTTPClient(httpClient *http.Client) RegistryClientOption

WithHTTPClient sets the HTTP client that executes requests, which is how you configure TLS, proxies, and timeouts.

func WithMaxRetries source

func WithMaxRetries(maxRetries int) RegistryClientOption

WithMaxRetries sets how many times a failed request is retried; zero disables retries. The default is two retries.

type RegistryError source

type RegistryError struct {
    // StatusCode is the HTTP status code, or -1 for a transport or parsing
    // error.
    StatusCode int
    // contains filtered or unexported fields
}

RegistryError reports a schema registry transport, status, or response error.

Returned by RegistryClient when a request cannot be sent, the registry answers with a non-2xx status, or the response body cannot be parsed. StatusCode is the HTTP status, or -1 when no status was received.

func (*RegistryError) Error source

func (e *RegistryError) Error() string

Error returns a description of the failure.

func (*RegistryError) Unwrap source

func (e *RegistryError) Unwrap() error

Unwrap returns the underlying error, or nil.

type ResolvedSchema source

type ResolvedSchema struct {
    // Schema is the schema text.
    Schema string

    // MessageType is the Protobuf message full name, or "" for other formats.
    MessageType string

    // References maps reference names to schema texts, in dependency-first
    // order of discovery.
    References map[string]string
}

ResolvedSchema is a schema fetched by ID with every transitive reference resolved to its text.

type Role source

type Role int

Role identifies whether a schema belongs to a record key or value.

The role selects the registry subject through the configured SubjectNameStrategy, for example "orders-key" versus "orders-value" under the default TopicNameStrategy. 

const (
    // RoleKey marks a schema that describes the record key.
    RoleKey Role = iota

    // RoleValue marks a schema that describes the record value.
    RoleValue
)

func (Role) String source

func (r Role) String() string

String returns "KEY" or "VALUE".

type SchemaCache source

type SchemaCache struct {
    // contains filtered or unexported fields
}

SchemaCache stores schema IDs and writer schemas for synchronous serde operations.

Serdes must never block a Kafka poll loop on registry I/O, so all registry traffic is funneled through this cache:

  1. Each serde calls SchemaCache.Intern (through its RegisterSubject method) to declare the subjects it will use.
  2. The application calls SchemaCache.Prewarm once at startup, which resolves every interned subject to a schema ID with the configured RegisterMode.
  3. Serialization then reads IDs synchronously with SchemaCache.IDForSubject, and deserialization reads writer schemas with SchemaCache.WriterSchema; an unknown ID starts one background fetch and returns the retriable FetchPendingError.

The cache is safe for concurrent use and can be shared by any number of serdes; sharing one cache per application is the intended usage.

Example (seeding)

Seeding the cache makes serdes fully deterministic: no registry is ever contacted, so the unreachable URL is intentional. 

package main

import (
    "fmt"

    "github.com/krabka-io/krabka-streams-go/schema"
)

func main() {
    client, err := schema.NewRegistryClient("http://127.0.0.1:1")
    if err != nil {
        panic(err)
    }
    cache := schema.NewSchemaCache(client)

    type Order struct {
        ID string `json:"id"`
    }
    schemaText := `{"type":"object","properties":{"id":{"type":"string"}},"required":["id"]}`
    serde, err := schema.NewJSONSchemaSerde[Order](schemaText, cache, schema.RoleValue, true)
    if err != nil {
        panic(err)
    }
    cache.SeedSubjectID("orders-value", 11)
    cache.SeedWriterSchema(11, schemaText)

    data, err := serde.Serialize("orders", Order{ID: "o-1"})
    if err != nil {
        panic(err)
    }
    back, err := serde.Deserialize("orders", data)
    if err != nil {
        panic(err)
    }
    fmt.Println(back.ID)
}

Output:

o-1

func NewSchemaCache source

func NewSchemaCache(client *RegistryClient, options ...SchemaCacheOption) *SchemaCache

NewSchemaCache creates a cache over the given registry client. Without options it auto-registers schemas and uses the topic naming rule.

func (*SchemaCache) IDForSubject source

func (c *SchemaCache) IDForSubject(subject string) (int, bool)

IDForSubject returns the resolved schema ID for a subject, if prewarming has resolved it.

func (*SchemaCache) Intern source

func (c *SchemaCache) Intern(subject string, kind Kind, schemaText, messageType string)

Intern adds a local schema to the next prewarm operation. This operation is idempotent by subject: interning the same subject twice keeps the first schema.

Serdes call this through their RegisterSubject methods; call it directly only when prewarming subjects that no local serde owns. messageType is the Protobuf message full name and "" for other formats.

func (*SchemaCache) Prewarm source

func (c *SchemaCache) Prewarm(ctx context.Context) error

Prewarm resolves all interned subject IDs with the configured registration mode.

All subjects are resolved concurrently. The returned error joins every per-subject failure; use SchemaCache.PrewarmReport for per-subject outcomes. Successfully resolved subjects are cached even when others fail.

func (*SchemaCache) PrewarmReport source

func (c *SchemaCache) PrewarmReport(ctx context.Context) PrewarmReport

PrewarmReport resolves every interned subject and reports successes and failures independently.

Unlike SchemaCache.Prewarm, it never fails as a whole; inspect the report to find out which subjects failed and why.

func (*SchemaCache) SeedSubjectID source

func (c *SchemaCache) SeedSubjectID(subject string, schemaID int)

SeedSubjectID adds a subject ID directly, without I/O. This method supports deterministic tests and offline startup.

func (*SchemaCache) SeedWriterMessageType source

func (c *SchemaCache) SeedWriterMessageType(schemaID int, messageType string)

SeedWriterMessageType adds Protobuf message metadata directly, without I/O.

func (*SchemaCache) SeedWriterSchema source

func (c *SchemaCache) SeedWriterSchema(schemaID int, schemaText string)

SeedWriterSchema adds a writer schema directly, without I/O. This method supports deterministic tests and offline startup.

func (*SchemaCache) Subject source

func (c *SchemaCache) Subject(topic string, role Role) string

Subject maps a topic and role to a subject with the cache's default strategy.

func (*SchemaCache) SubjectWith source

func (c *SchemaCache) SubjectWith(topic string, role Role, strategy SubjectNameStrategy) string

SubjectWith maps a topic and role to a subject with an explicit strategy instead of the cache default.

func (*SchemaCache) WriterMessageType source

func (c *SchemaCache) WriterMessageType(schemaID int) string

WriterMessageType returns the cached Protobuf message full name for a schema ID, or "" when unknown or not Protobuf.

func (*SchemaCache) WriterReferences source

func (c *SchemaCache) WriterReferences(schemaID int) map[string]string

WriterReferences returns the cached referenced schemas for a schema ID as a map of reference names to schema texts, empty when the schema has no references.

func (*SchemaCache) WriterSchema source

func (c *SchemaCache) WriterSchema(schemaID int) (string, error)

WriterSchema returns a cached writer schema. A cache miss starts one background fetch and returns a retriable *FetchPendingError.

This method never blocks. Only one fetch per schema ID runs at a time; concurrent callers for the same ID all receive the pending error until the fetch completes. If the background fetch fails, its marker is removed and the next call starts a fresh fetch.

type SchemaCacheOption source

type SchemaCacheOption func(*SchemaCache)

SchemaCacheOption configures a SchemaCache.

func WithRegisterMode source

func WithRegisterMode(mode RegisterMode) SchemaCacheOption

WithRegisterMode sets how prewarming resolves each interned subject to an ID. The default is AutoRegister.

func WithSubjectNameStrategy source

func WithSubjectNameStrategy(strategy SubjectNameStrategy) SchemaCacheOption

WithSubjectNameStrategy sets the default topic-to-subject mapping. The default is TopicNameStrategy.

type SchemaReference source

type SchemaReference struct {
    // Name is the name the referencing schema uses, for example an Avro full
    // name or a Protobuf import path.
    Name string `json:"name"`

    // Subject is the referenced subject.
    Subject string `json:"subject"`

    // Version is the referenced version within that subject.
    Version int `json:"version"`
}

SchemaReference is a reference from one schema to a version of another subject.

type Serde source

type Serde[T any] interface {
    // Serialize encodes a value for a topic into Confluent-framed bytes.
    Serialize(topic string, value T) ([]byte, error)

    // Deserialize decodes Confluent-framed bytes read from a topic.
    Deserialize(topic string, data []byte) (T, error)

    // RegisterSubject interns the serde's schema under the topic's subject
    // for the next prewarm. It performs no I/O and is idempotent per subject.
    RegisterSubject(topic string)
}

Serde produces and consumes Confluent-framed record bytes for one record role.

All serdes in this package share the same lifecycle: construct the serde for a fixed Role, call RegisterSubject for every topic it will serve, prewarm the shared SchemaCache, and then serialize and deserialize synchronously. Serialization and deserialization never perform I/O.

A nil-like value (a nil pointer, map, slice, or interface) serializes to nil bytes, so Kafka tombstones survive untouched, and nil bytes deserialize to the zero value.

type SerdeOption source

type SerdeOption func(*serdeOptions)

SerdeOption configures serde construction.

func WithStrategy source

func WithStrategy(strategy SubjectNameStrategy) SerdeOption

WithStrategy sets the subject name strategy the serde uses instead of the cache default. Each serde accepts its own strategy, so one cache can serve topic-name and record-name subjects together.

type SubjectNameStrategy source

type SubjectNameStrategy func(topic string, role Role) string

SubjectNameStrategy maps a topic and record role to a schema registry subject.

The default strategy is TopicNameStrategy. Provide a custom implementation to serdes or to SchemaCache when subjects follow a different convention, such as one subject per record type shared across topics.