Package sr provides a schema registry client and a helper type to encode values and decode data according to the schema registry wire format.
As mentioned on the Serde type, this package does not provide schema auto-discovery and type auto-decoding. To aid in strong typing and validated encoding/decoding, you must register IDs and values to how to encode or decode them.
The client does not automatically cache schemas, instead, the Serde type is used for the actual caching of IDs to how to encode/decode the IDs. The Client type itself simply speaks http to your schema registry and returns the results.
To read more about the schema registry, see the following:
https://docs.confluent.io/platform/current/schema-registry/develop/api.html
const ( LogLevelNone int8 = iota LogLevelError LogLevelWarn LogLevelInfo LogLevelDebug )
The log levels that can be used to control the verbosity of the SR client. The levels are ordered to mirror the kgo leveling.
GlobalSubject is a constant to make API usage of requesting global subjects clearer.
View Sourcevar (
ErrUnknown = &Error{-1, "UNKNOWN_SCHEMA_REGISTRY_ERROR", "Unknown Schema Registry error"}
ErrSubjectNotFound = &Error{40401, "SUBJECT_NOT_FOUND", "Subject does not exist"}
ErrVersionNotFound = &Error{40402, "VERSION_NOT_FOUND", "Version does not exist for subject"}
ErrSchemaNotFound = &Error{40403, "SCHEMA_NOT_FOUND", "Schema does not exist"}
ErrSubjectSoftDeleted = &Error{40404, "SUBJECT_SOFT_DELETED", "Subject was soft deleted"}
ErrSubjectNotSoftDeleted = &Error{40405, "SUBJECT_NOT_SOFT_DELETED", "Subject was not deleted first before being permanently deleted"}
ErrSchemaVersionSoftDeleted = &Error{40406, "SCHEMA_VERSION_SOFT_DELETED", "Schema version was soft deleted"}
ErrSchemaVersionNotSoftDeleted = &Error{40407, "SCHEMA_VERSION_NOT_SOFT_DELETED", "Schema version was not deleted first before being permanently deleted"}
ErrSubjectLevelCompatibilityNotConfigured = &Error{40408, "SUBJECT_LEVEL_COMPATIBILITY_NOT_CONFIGURED", "Subject does not have subject-level compatibility configured"}
ErrSubjectLevelModeNotConfigured = &Error{40409, "SUBJECT_LEVEL_MODE_NOT_CONFIGURED", "Subject does not have subject-level mode configured"}
ErrIncompatibleSchema = &Error{409, "INCOMPATIBLE_SCHEMA", "Schema is incompatible with an earlier schema"}
ErrInvalidSchema = &Error{42201, "INVALID_SCHEMA", "Schema is invalid"}
ErrInvalidVersion = &Error{42202, "INVALID_VERSION", "Version is invalid"}
ErrInvalidCompatibilityLevel = &Error{42203, "INVALID_COMPATIBILITY_LEVEL", "Compatibility level is invalid"}
ErrInvalidMode = &Error{42204, "INVALID_MODE", "Mode is invalid"}
ErrOperationNotPermitted = &Error{42205, "OPERATION_NOT_PERMITTED", "Operation is not permitted"}
ErrReferenceExists = &Error{42206, "REFERENCE_EXISTS", "Schema reference already exists"}
ErrIDDoesNotMatch = &Error{42207, "ID_DOES_NOT_MATCH", "Provided ID does not match expected ID"}
ErrInvalidSubject = &Error{42208, "INVALID_SUBJECT", "Subject name is invalid"}
ErrSchemaTooLarge = &Error{42209, "SCHEMA_TOO_LARGE", "Schema is too large"}
ErrInvalidRuleset = &Error{42210, "INVALID_RULESET", "Ruleset is invalid"}
ErrContextNotEmpty = &Error{42211, "CONTEXT_NOT_EMPTY", "Context is not empty when it should be"}
ErrStoreError = &Error{50001, "STORE_ERROR", "Error in backend datastore"}
ErrOperationTimeout = &Error{50002, "OPERATION_TIMEOUT", "Operation timed out"}
ErrRequestForwardingFailed = &Error{50003, "REQUEST_FORWARDING_FAILED", "Error forwarding request to primary"}
ErrUnknownLeader = &Error{50004, "UNKNOWN_LEADER", "Leader is unknown"}
ErrJSONParseError = &Error{50005, "JSON_PARSE_ERROR", "JSON parsing error"}
)
Schema Registry errors as defined in the official Confluent Schema Registry. These allow clients to programmatically check for specific error conditions when interacting with the Schema Registry API.
The errors are organized by HTTP status code categories:
var (
ErrNotRegistered = errors.New("registration is missing for encode/decode")
ErrBadHeader = errors.New("5 byte header for value is missing or does not have 0 magic byte")
)
AppendEncode encodes a value and prepends the header, appends it to b and returns b. If the encoding function fails, this returns an error.
Encode encodes a value and prepends the header. If the encoding function fails, this returns an error.
func IsInvalidRequestError(code int) bool
IsInvalidRequestError returns true if the error code indicates an invalid request.
func IsNotFoundError(code int) bool
IsNotFoundError returns true if the error code indicates a "not found" condition.
IsServerError returns true if the error code indicates a server-side error.
WithParams adds query parameters to the given context. This is a merge operation: any non-zero parameter is kept. The variadic nature of this allows for a nicer api:
sr.WithParams(ctx, sr.Format("default"), sr.FetchMaxID)
type CheckCompatibilityResult struct {
Is bool `json:"is_compatible"`
Messages []string `json:"messages"`
}
CheckCompatibilityResult is the response from the check compatibility endpoint.
Client talks to a schema registry and contains helper functions to serialize and deserialize objects according to schemas.
NewClient returns a new schema registry client.
CheckCompatibility checks if a schema is compatible with the given version that exists. You can use -1 to check compatibility with the latest version, and -2 to check compatibility against all versions.
This supports params Normalize and Verbose.
Compatibility returns the subject compatibility and global compatibility of each requested subject. The global compatibility can be requested by using either an empty subject or by specifying no subjects.
This supports params DefaultToGlobal.
This can return 200 or 500 per result.
CreateSchema attempts to create a schema in the given subject.
If you want to register or create a schema with a specific ID and/or version, you can use [RegisterSchema] or [CreateSchemaWithIDAndVersion].
This supports param Normalize.
func (*Client) CreateSchemaWithIDAndVersion ΒΆ added in v1.3.0CreateSchemaWithIDAndVersion attempts to create a schema with a fixed ID and version in the given subject. If the id is set to -1 or 0, this method is equivalent to [CreateSchema]. If the version is set to -1 or 0, the registry will interpret it as "latest".
This function first registers the schema and then looks up all details. If you want to register a schema, only receive the registered ID back, and avoid further HTTP requests to the registry, you can use [RegisterSchema] instead.
This supports param Normalize.
DeleteSchema deletes the schema at the given version. You must soft delete a schema before it can be hard deleted. You can use -1 to delete the latest version.
DeleteSubject deletes the subject. You must soft delete a subject before it can be hard deleted. This returns all versions that were deleted.
Do sends an HTTP request to the schema registry using the given method and path, optionally encoding the provided request body `v` as JSON, and decoding the response into `into` if non-nil.
This method is a general-purpose extension point for users who need to interact with custom or non-standard schema registry endpoints while reusing the client's configured authentication, user-agent, retry logic, and error handling.
LookupSchema checks to see if a schema is already registered and if so, returns its ID and version in the SubjectSchema.
This supports params Normalize and Deleted.
Mode returns the subject and global mode of each requested subject. The global mode can be requested by using either an empty subject or by specifying no subjects.
This supports params DefaultToGlobal.
OptValue returns the value for the given configuration option. If the given option does not exist, this returns nil. This function takes either a raw ClientOpt, or an Opt function name.
If a configuration option has multiple inputs, this function returns only the first input. Variadic option inputs are returned as a single slice. Options that are internally stored as a pointer are returned as their string input; you can see if the option is internally nil by looking at the second value returned from OptValues.
var (
cl, _ := NewClient(
URLs("foo", "bar"),
UserAgent("baz"),
)
urls = cl.OptValue("URLs") // urls is []string{"foo", "bar"}; string lookup for the option works
ua = cl.OptValue(UserAgent) // ua is "baz"
unk = cl.OptValue("Unknown"), // unk is nil
)
OptValues returns all values for options. This method is useful for options that have multiple inputs (notably, BasicAuth). This is also useful for options that are internally stored as a pointer -- this function will return the string value of the option but also whether the option is non-nil. Boolean options are returned as a single-element slice with the bool value. Variadic inputs are returned as a signle slice. If the input option does not exist, this returns nil.
var (
cl, _ := NewClient(
URLs("foo", "bar"),
UserAgent("baz"),
)
urls = cl.OptValues("URLs") // urls is []any{[]string{"foo", "bar"}}
ua = cl.OptValues(UserAgent) // ua is []any{"baz"}
ba = cl.OptValues(BasicAuth) // ba is []any{"user", "pass"}
unk = cl.OptValues("Unknown"), // unk is nil
)
Opts returns the options that were used to create this client. This can be as a base to generate a new client, where you can add override options to the end of the original input list.
RegisterSchema attempts to create a schema with a fixed ID and version in the given subject, and returns the globally unique identifier assigned by the registry. If the id is set to -1 or 0, the registry will try to allocate a new identifier or reuse an existing one. If the version is set to -1 or 0, the registry will interpret it as "latest".
If the schema is already registered, the original ID will be returned and a new version will not be created.
If you want to register a schema and receive more details back, you can use [CreateSchema] or [CreateSchemaWithIDAndVersion] at the expense of more HTTP requests to the registry.
This supports param Normalize.
ResetCompatibility deletes any subject-level compatibility and reverts to the global default. The global compatibility can be reset by either using an empty subject or by specifying no subjects.
This can return 200 or 500.
ResetMode deletes any subject modes and reverts to the global default.
SchemaByVersion returns the schema for a given subject and version. You can use -1 as the version to return the latest schema.
This supports param ShowDeleted.
SchemaReferences returns all schemas that references the input subject-version. You can use -1 to check the latest version.
This supports param ShowDeleted.
SchemaTextByID returns the actual text of a schema.
For example, if the schema for an ID is
"{\"type\":\"boolean\"}"
this will return
{"type":"boolean"}
This supports params Subject, Format.
SchemaTextByVersion returns the actual text of a schema, by subject and version. You can use -1 as the version to return the latest schema.
For example, if the schema for an ID is
"{\"type\":\"boolean\"}"
this will return
{"type":"boolean"}
This supports param ShowDeleted.
SchemaUsagesByID returns all usages of a given schema ID. A single schema's can be reused in many subject-versions; this function can be used to map a schema to all subject-versions that use it.
This supports param ShowDeleted.
SchemaVersionsByID returns all subject versions associated with a schema ID.
This supports params Subject and ShowDeleted.
Schemas returns all schemas for the given subject.
This supports param ShowDeleted.
SetCompatibility sets the compatibility for each requested subject. The global compatibility can be set by either using an empty subject or by specifying no subjects. If specifying no subjects, this returns one element.
SetMode sets the mode for each requested subject. The global mode can be set by either using an empty subject or by specifying no subjects. If specifying no subjects, this returns one element.
This supports params Force.
SubjectsByID returns the subjects associated with a schema ID.
This supports params Subject and ShowDeleted.
SupportedTypes returns the schema types that are supported in the schema registry.
type ClientOpt interface {
}
ClientOpt is an option to configure a client.
BasicAuth sets basic authorization to use for every request.
BearerToken sets an Authorization header to use for every request. The format will be: "Authorization: Bearer $token".
DefaultParams sets default parameters to apply to every request.
DialTLSConfig sets a tls.Config to use in the http client, either the default client or a client previously set with the HTTPClient option. When setting this option, HTTP/2 support may not be enabled by default.
HTTPClient sets the http client that the schema registry client uses, overriding the default client that speaks plaintext with a timeout of 5s.
LogFn sets the logger function to use.
LogLevel sets a static log level to use, overriding the default "info" level.
There are five levels:
This package defines int8 constants for convenience. The levels (and log function) mirror kgo's Logger and LogLevel definitions, making it easy to use any existing kgo logging functionality you may already be using in this package as well.
LogLevelFn sets a function to return the log level dynamically. See LogLevel for more information.
PreReq sets a hook func to call before every request is sent.
URLs sets the URLs that the client speaks to, overriding the default http://localhost:8081. This option automatically prefixes any URL that is missing an http:// or https:// prefix with http://.
UserAgent sets the User-Agent to use in requests, overriding the default "franz-go".
type CompatibilityLevel int
CompatibilityLevel as an enum representing config compatibility levels.
const ( CompatNone CompatibilityLevel = 1 + iota CompatBackward CompatBackwardTransitive CompatForward CompatForwardTransitive CompatFull CompatFullTransitive )
CompatibilityResult is the compatibility level for a subject.
type ConfluentHeader struct{}
ConfluentHeader is a SerdeHeader that produces the Confluent wire format. It starts with 0, then big endian uint32 of the ID, then index (only protobuf), then the encoded message.
https://docs.confluent.io/platform/current/schema-registry/serdes-develop/index.html#wire-format
AppendEncode appends an encoded header to b according to the Confluent wire format and returns it. Error is always nil.
DecodeID strips and decodes the schema ID from b. It returns the ID alongside the unread bytes. If the header does not contain the magic byte or b contains less than 5 bytes it returns ErrBadHeader.
DecodeIndex strips and decodes indices from b. It returns the index slice alongside the unread bytes. It expects b to be the output of DecodeID (schema ID should already be stripped away). If maxLength is greater than 0 and the encoded data contains more indices than maxLength the function returns ErrNotRegistered.
UpdateID replaces the schema ID in b. If the header does not contain the magic byte or b contains less than 5 bytes it returns ErrBadHeader.
DeleteHow is a typed bool indicating how subjects or schemas should be deleted.
type EncodingOpt interface {
}
EncodingOpt is an option to configure the behavior of Serde.Encode and Serde.Decode.
AppendEncodeFn allows Serde to encode a value to an existing slice. This can be more efficient than EncodeFn; this function is used if it exists.
DecodeFn allows Serde to decode into a value.
EncodeFn allows Serde to encode a value.
GenerateFn returns a new(Value) that can be decoded into. This function can be used to control the instantiation of a new type for DecodeNew.
Index attaches a message index to a value. A single schema ID can be registered multiple times with different indices.
This option supports schemas that encode many different values from the same schema (namely, protobuf). The index into the schema to encode a particular message is specified with `index`.
NOTE: this option must be used for protobuf schemas.
For more information, see where `message-indexes` are described in:
https://docs.confluent.io/platform/current/schema-registry/serdes-develop/index.html#wire-format
Error represents a Schema Registry error with its code, name, and description.
ErrorForCode returns the Error corresponding to the given error code. If the code is unknown, this returns ErrUnknown.
Error implements the error interface, returning the human-readable description.
LogValue implements slog.LogValuer to provide structured logging output.
Mode as an enum representing the "mode" of the registry or a subject.
const ( ModeImport Mode = iota ModeReadOnly ModeReadWrite )
ModeResult is the mode for a subject.
Param is a parameter that can be passed to various APIs. Each API documents the parameters they accept.
Format returns a Param that configures how schema's are returned in certain get-schema operations.
For Avro schemas, the Format param supports "default" or "resolved". For Protobuf schemas, the Format param supports "default", "ignore_extensions", or "serialized".
RawParams returns a Param with raw query parameters.
Subject returns a Param limiting which subject is returned in certain list-schema or list-subject operations.
SubjectPrefix returns a Param that filters subjects by prefix when listing schemas.
type ResponseError struct {
Method string `json:"-"`
URL string `json:"-"`
StatusCode int `json:"-"`
Raw []byte `json:"-"`
ErrorCode int `json:"error_code"`
Message string `json:"message"`
}
ResponseError is the type returned from the schema registry for errors.
SchemaError returns the typed Error corresponding to this response error's error code. If the error code is unknown, this returns nil.
Schema is the object form of a schema for the HTTP API.
type SchemaMetadata struct {
Tags map[string][]string `json:"tags,omitempty"`
Properties map[string]string `json:"properties,omitempty"`
Sensitive []string `json:"sensitive,omitempty"`
}
SchemaMetadata is arbitrary information about the schema or its constituent parts, such as whether a field contains sensitive information or who created a data contract.
type SchemaReference struct {
Name string `json:"name"`
Subject string `json:"subject"`
Version int `json:"version"`
}
SchemaReference is a way for a one schema to reference another. The details for how referencing is done are type specific; for example, JSON objects that use the key "$ref" can refer to another schema via URL. For more details on references, see the following link:
https://docs.confluent.io/platform/current/schema-registry/serdes-develop/index.html#schema-references https://docs.confluent.io/platform/current/schema-registry/develop/api.html
SchemaRule specifies integrity constraints or data policies in a data contract. These data rules or policies can enforce that a field that contains sensitive information must be encrypted, or that a message containing an invalid age must be sent to a dead letter queue
https://docs.confluent.io/platform/current/schema-registry/fundamentals/data-contracts.html#rules
SchemaRuleKind as an enum representing the kind of schema rule.
SchemaRuleMode specifies a schema rule's mode.
Migration rules can be specified for an UPGRADE, DOWNGRADE, or both (UPDOWN). Migration rules are used during complex schema evolution.
Domain rules can be specified during serialization (WRITE), deserialization (READ) or both (WRITEREAD).
Domain rules can be used to transform the domain values in a message payload.
const ( SchemaRuleModeUpgrade SchemaRuleMode = iota SchemaRuleModeDowngrade SchemaRuleModeUpdown SchemaRuleModeWrite SchemaRuleModeRead SchemaRuleModeWriteRead )
type SchemaRuleSet struct {
MigrationRules []SchemaRule `json:"migrationRules,omitempty"`
DomainRules []SchemaRule `json:"domainRules,omitempty"`
}
SchemaRuleSet groups migration rules and domain validation rules.
SchemaType as an enum representing schema types. The default schema type is avro.
Serde encodes and decodes values according to the schema registry wire format. A Serde itself does not perform schema auto-discovery and type auto-decoding. To aid in strong typing and validated encoding/decoding, you must register IDs and values to how to encode or decode them.
To use a Serde for encoding, you must pre-register schema ids and values you will encode, and then you can use the encode functions.
To use a Serde for decoding, you can either pre-register schema ids and values you will consume, or you can discover the schema every time you receive an ErrNotRegistered error from decode.
NewSerde returns a new Serde using the supplied default options, which are applied to every registered type. These options are always applied first, so you can override them as necessary when registering.
This can be useful if you always want to use the same encoding or decoding functions.
AppendEncode encodes a value and prepends the header according to the configured SerdeHeader, appends it to b and returns b. If EncodeFn was not registered, this returns ErrNotRegistered.
Decode decodes b into v. If DecodeFn option was not used, this returns ErrNotRegistered.
Serde does not handle references in schemas; it is up to you to register the full decode function for any top-level ID, regardless of how many other schemas are referenced in top-level ID.
DecodeID decodes an ID from b, returning the ID and the remaining bytes, or an error.
DecodeIndex decodes at most maxLength of a schema index from in, returning the index and remaining bytes, or an error. It expects b to be the output of DecodeID (schema ID should already be stripped away).
DecodeNew is the same as Decode, but decodes into a new value rather than the input value. If DecodeFn was not used, this returns ErrNotRegistered. GenerateFn can be used to control the instantiation of a new value, otherwise this uses reflect.New(reflect.TypeOf(v)).Interface().
Encode encodes a value and prepends the header according to the configured SerdeHeader. If EncodeFn was not used, this returns ErrNotRegistered.
MustAppendEncode returns the value of AppendEncode, panicking on error. This is a shortcut for if your encode function cannot error.
MustEncode returns the value of Encode, panicking on error. This is a shortcut for if your encode function cannot error.
Register registers a schema ID and the value it corresponds to, as well as the encoding or decoding functions. You need to register functions depending on whether you are only encoding, only decoding, or both.
SerdeHeader encodes and decodes a message header.
type SerdeOpt interface {
}
SerdeOpt is an option to configure Serde.
Header defines the SerdeHeader used to encode and decode the message header.
type SerdeOrEncodingOpt interface {
}
SerdeOrEncodingOpt is either a SerdeOpt or EncodingOpt.
type SetCompatibility struct {
Level CompatibilityLevel `json:"compatibility"`
Alias string `json:"alias,omitempty"`
Normalize bool `json:"normalize,omitempty"`
Group string `json:"compatibilityGroup,omitempty"`
DefaultMetadata *SchemaMetadata `json:"defaultMetadata,omitempty"`
OverrideMetadata *SchemaMetadata `json:"overrideMetadata,omitempty"`
DefaultRuleSet *SchemaRuleSet `json:"defaultRuleSet,omitempty"`
OverrideRuleSet *SchemaRuleSet `json:"overrideRuleSet,omitempty"`
}
SetCompatibility contains information used for setting global or per-subject compatibility configuration.
The main difference between this and the CompatibilityResult is that this struct marshals the compatibility level as "compatibility".
type SubjectSchema struct {
Subject string `json:"subject,omitempty"`
Version int `json:"version,omitempty"`
ID int `json:"id,omitempty"`
Schema
}
SubjectSchema pairs the subject, global identifier, and version of a schema with the schema itself.
CommSubjectSchemas splits l and r into three sets: what is unique in l, what is unique in r, and what is common in both. Duplicates in either map are eliminated.
type SubjectVersion struct {
Subject string `json:"subject"`
Version int `json:"version"`
}
SubjectVersion is a subject version pair.
RetroSearch is an open source project built by @garambo | Open a GitHub Issue
Search and Browse the WWW like it's 1997 | Search results from DuckDuckGo
HTML:
3.2
| Encoding:
UTF-8
| Version:
0.7.4