Cloud

schema_registry_decode

Available in: Cloud, Self-Managed

Automatically decodes and validates messages with schemas from a Confluent Schema Registry service.

This processor uses the Franz Kafka Schema Registry client.

Common
Advanced

processors:
  label: ""
  schema_registry_decode:
    avro:
      raw_unions: "" # No default (optional)
      preserve_logical_types: false
      translate_kafka_connect_types: false
      mapping: "" # No default (optional)
      store_schema_metadata: "" # No default (optional)
    protobuf:
      use_proto_names: false
      use_enum_numbers: false
      emit_unpopulated: false
      emit_default_values: false
      serialize_to_json: true
    cache_duration: 10m
    url: "" # No default (required)
    default_schema_id: "" # No default (optional)

processors:
  label: ""
  schema_registry_decode:
    avro:
      raw_unions: "" # No default (optional)
      preserve_logical_types: false
      translate_kafka_connect_types: false
      mapping: "" # No default (optional)
      store_schema_metadata: "" # No default (optional)
    protobuf:
      use_proto_names: false
      use_enum_numbers: false
      emit_unpopulated: false
      emit_default_values: false
      serialize_to_json: true
    cache_duration: 10m
    url: "" # No default (required)
    default_schema_id: "" # No default (optional)
    oauth:
      enabled: false
      consumer_key: ""
      consumer_secret: ""
      access_token: ""
      access_token_secret: ""
    basic_auth:
      enabled: false
      username: ""
      password: ""
    jwt:
      enabled: false
      private_key_file: ""
      signing_method: ""
      claims: {}
      headers: {}
    tls:
      skip_cert_verify: false
      enable_renegotiation: false
      root_cas: ""
      root_cas_file: ""
      client_certs: []

Decodes messages automatically from a schema stored within a Confluent Schema Registry service by extracting a schema ID from the message and obtaining the associated schema from the registry. If a message fails to match against the schema then it will remain unchanged and the error can be caught using error-handling methods.

Avro, Protobuf and JSON schemas are supported, all are capable of expanding from schema references as of v4.22.0.

Avro JSON format

By default, this processor expects documents formatted as Avro JSON when decoding with Avro schemas. In this format, the value of a union is encoded in JSON as follows:

If the union’s type is null, it is encoded as a JSON null.
Otherwise, the union is encoded as a JSON object with one name/value pair. The name is the type’s name, and the value is the recursively-encoded value. The user-specified name is used for Avro’s named types (record, fixed, or enum). For other types, the type name is used.

For example, the union schema ["null","string","Transaction"], where Transaction is a record name, would encode:

null as a JSON null
The string "a" as {"string": "a"}
A Transaction instance as {"Transaction": {…}}, where {…} indicates the JSON encoding of a Transaction instance

Alternatively, you can create documents in standard/raw JSON format by setting the field avro.raw_unions to true.

Protobuf format

This processor decodes Protobuf messages to JSON documents. For more information about the JSON mapping of Protobuf messages, see the Protocol Buffers documentation.

Metadata

This processor adds the following metadata to processed messages:

schema_id: The ID of the schema in the schema registry associated with the message.

Fields

`avro`

Configuration for how to decode schemas that are of type AVRO.

Type: object

`avro.mapping`

Define a custom mapping to apply to the JSON representation of Avro schemas. You can use mappings to convert custom types emitted by other tools, such as Debezium, into standard Avro types.

Type: string

# Examples:
mapping: |-

  map isDebeziumTimestampType {
    root = this.type == "long" && this."connect.name" == "io.debezium.time.Timestamp" && !this.exists("logicalType")
  }
  map debeziumTimestampToAvroTimestamp {
    let mapped_fields = this.fields.or([]).map_each(item -> item.apply("debeziumTimestampToAvroTimestamp"))
    root = match {
      this.type == "record" => this.assign({"fields": $mapped_fields})
      this.type.type() == "array" => this.assign({"type": this.type.map_each(item -> item.apply("debeziumTimestampToAvroTimestamp"))})
      # Add a logical type so that it's decoded as a timestamp instead of a long.
      this.type.type() == "object" && this.type.apply("isDebeziumTimestampType") => this.merge({"type":{"logicalType": "timestamp-millis"}})
      _ => this
    }
  }
  root = this.apply("debeziumTimestampToAvroTimestamp")

`avro.preserve_logical_types`

Choose whether to:

Transform logical types into their primitive type (default). For example, decimals become raw bytes and timestamps become plain integers.
Preserve logical types.

Set to true to preserve logical types.

Type: bool

Default: false

`avro.raw_unions`

Whether Avro messages should be decoded into normal JSON (JSON that meets the expectations of regular internet JSON) rather than Avro JSON.

If set to false, Avro messages are decoded as Avro JSON.

For example, the union schema ["null","string","Transaction"], where Transaction is a record name, would be decoded as:

A null as a JSON null
The string "a" as {"string": "a"}
A Transaction instance as {"Transaction": {…}}, where {…} indicates the JSON encoding of a Transaction instance.

If set to true, Avro messages are decoded as standard JSON.

For example, the same union schema ["null","string","Transaction"] is decoded as:

A null as JSON null
The string "a" as "a"
A Transaction instance as {…}, where {…} indicates the JSON encoding of a Transaction instance.

For more details on the difference between standard JSON and Avro JSON, see the comment in Goavro and the underlying library used for Avro serialization.

Type: bool

`avro.store_schema_metadata`

Optionally store the schema used to decode messages as a metadata field under the given name. This field can later be referenced in other components such as a parquet_encode processor in order to automatically infer their schema.

Type: string

`avro.translate_kafka_connect_types`

Only valid if preserve_logical_types is true. This decodes various Kafka Connect types into their bloblang equivalents when not representable by standard logical types according to the Avro standard.

Types that are currently translated:

Table 1. Debezium Custom Temporal Types
Type Name	Bloblang Type	Description
io.debezium.time.Date	timestamp	Date without time (days since epoch)
io.debezium.time.Timestamp	timestamp	Timestamp without timezone (milliseconds since epoch)
io.debezium.time.MicroTimestamp	timestamp	Timestamp with microsecond precision
io.debezium.time.NanoTimestamp	timestamp	Timestamp with nanosecond precision
io.debezium.time.ZonedTimestamp	timestamp	Timestamp with timezone (ISO-8601 format)
io.debezium.time.Year	timestamp at January 1st at 00:00:00	Year value
io.debezium.time.Time	timestamp at the unix epoch	Time without date (milliseconds past midnight)
io.debezium.time.MicroTime	timestamp at the unix epoch	Time with microsecond precision
io.debezium.time.NanoTime	timestamp at the unix epoch	Time with nanosecond precision

Type: bool

Default: false

`basic_auth`

Allows you to specify basic authentication.

Type: object

`basic_auth.enabled`

Whether to use basic authentication in requests.

Type: bool

Default: false

`basic_auth.password`

A password to authenticate with.

This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see Manage Secrets before adding it to your configuration.

Type: string

Default: ""

`basic_auth.username`

A username to authenticate as.

Type: string

Default: ""

`cache_duration`

The duration after which a cached schema is considered stale and is removed from the cache.

Type: string

Default: 10m

# Examples:
cache_duration: 1h

# ---

cache_duration: 5m

`default_schema_id`

This schema ID is used when a message’s schema header cannot be read (ErrBadHeader). If this value is not set, schema header errors are returned.

This configuration does not work with protobuf schemas.

You can also use the with_schema_registry_header bloblang function to add a schema ID to messages.

Type: int

`jwt`

Beta

Configure JSON Web Token (JWT) authentication. This feature is in beta and may change in future releases. JWT tokens provide secure, stateless authentication between services.

Type: object

`jwt.claims`

A value used to identify the claims that issued the JWT.

Type: object

Default: {}

`jwt.enabled`

Whether to use JWT authentication in requests.

Type: bool

Default: false

`jwt.headers`

Additional key-value pairs to include in the JWT header (optional). These headers provide extra metadata for JWT processing.

Type: object

Default: {}

`jwt.private_key_file`

Path to a file containing the PEM-encoded private key using PKCS#1 or PKCS#8 format. The private key must be compatible with the algorithm specified in the signing_method field.

Type: string

Default: ""

`jwt.signing_method`

The cryptographic algorithm used to sign the JWT token. Supported algorithms include RS256, RS384, RS512, and EdDSA. This algorithm must be compatible with the private key specified in the private_key_file field.

Type: string

Default: ""

`oauth`

Configure OAuth version 1.0 authentication for secure API access.

Type: object

`oauth.access_token`

A value used to gain access to the protected resources on behalf of the user.

Type: string

Default: ""

`oauth.access_token_secret`

A secret provided in order to establish ownership of a given access token.

This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see Manage Secrets before adding it to your configuration.

Type: string

Default: ""

`oauth.consumer_key`

A value used to identify the client to the service provider.

Type: string

Default: ""

`oauth.consumer_secret`

A secret used to establish ownership of the consumer key.

This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see Manage Secrets before adding it to your configuration.

Type: string

Default: ""

`oauth.enabled`

Whether to use OAuth version 1 in requests.

Type: bool

Default: false

`protobuf`

Configuration for how to decode schemas that are of type PROTOBUF.

Type: object

`protobuf.emit_default_values`

Whether to emit default-valued primitive fields, empty lists, and empty maps. emit_unpopulated takes precedence over emit_default_values

Type: bool

Default: false

`protobuf.emit_unpopulated`

Whether to emit unpopulated fields. It does not emit unpopulated oneof fields or unpopulated extension fields.

Type: bool

Default: false

`protobuf.serialize_to_json`

If messages should be serialized to JSON bytes. If false then the message is kept in decoded form, which means that 64 bit integers are not converted to strings and types for bytes and google.protobuf.Timestamp are preserved (as they are not serialized to JSON strings).

Type: bool

Default: true

`protobuf.use_enum_numbers`

Emits enum values as numbers.

Type: bool

Default: false

`protobuf.use_proto_names`

Use proto field name instead of lowerCamelCase name.

Type: bool

Default: false

`tls`

Custom TLS settings can be used to override system defaults.

Type: object

`tls.client_certs[]`

A list of client certificates to use. For each certificate either the fields cert and key, or cert_file and key_file should be specified, but not both.

Type: object

Default: []

# Examples:
client_certs:
  - cert: foo
    key: bar


# ---

client_certs:
  - cert_file: ./example.pem
    key_file: ./example.key

`tls.client_certs[].cert`

A plain text certificate to use.

Type: string

Default: ""

`tls.client_certs[].cert_file`

The path of a certificate to use.

Type: string

Default: ""

`tls.client_certs[].key`

A plain text certificate key to use.

This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see Manage Secrets before adding it to your configuration.

Type: string

Default: ""

`tls.client_certs[].key_file`

The path of a certificate key to use.

Type: string

Default: ""

`tls.client_certs[].password`

A plain text password for when the private key is password encrypted in PKCS#1 or PKCS#8 format. The obsolete pbeWithMD5AndDES-CBC algorithm is not supported for the PKCS#8 format.

Because the obsolete pbeWithMD5AndDES-CBC algorithm does not authenticate the ciphertext, it is vulnerable to padding oracle attacks that can let an attacker recover the plaintext.

This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see Manage Secrets before adding it to your configuration.

Type: string

Default: ""

# Examples:
password: foo

# ---

password: ${KEY_PASSWORD}

`tls.enable_renegotiation`

Whether to allow the remote server to repeatedly request renegotiation. Enable this option if you’re seeing the error message local error: tls: no renegotiation.

Type: bool

Default: false

`tls.root_cas`

An optional root certificate authority to use. This is a string, representing a certificate chain from the parent trusted root certificate, to possible intermediate signing certificates, to the host certificate.

This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see Manage Secrets before adding it to your configuration.

Type: string

Default: ""

# Examples:
root_cas: |-
  -----BEGIN CERTIFICATE-----
  ...
  -----END CERTIFICATE-----

`tls.root_cas_file`

An optional path of a root certificate authority file to use. This is a file, often with a .pem extension, containing a certificate chain from the parent trusted root certificate, to possible intermediate signing certificates, to the host certificate.

Type: string

Default: ""

# Examples:
root_cas_file: ./root_cas.pem

`tls.skip_cert_verify`

Whether to skip server side certificate verification.

Type: bool

Default: false

`url`

The base URL of the schema registry service.

Type: string

Was this helpful?

group Ask in the community

mail Share your feedback

group_add Make a contribution

What do you think of this page?

Let us know more:

Let us contact you about your feedback:

schema_registry_decode

Avro JSON format

Protobuf format

Metadata

Fields

avro

avro.mapping

avro.preserve_logical_types

avro.raw_unions

avro.store_schema_metadata

avro.translate_kafka_connect_types

basic_auth

basic_auth.enabled

basic_auth.password

basic_auth.username

cache_duration

default_schema_id

jwt

jwt.claims

jwt.enabled

jwt.headers

jwt.private_key_file

jwt.signing_method

oauth

oauth.access_token

oauth.access_token_secret

oauth.consumer_key

oauth.consumer_secret

oauth.enabled

protobuf

protobuf.emit_default_values

protobuf.emit_unpopulated

protobuf.serialize_to_json

protobuf.use_enum_numbers

protobuf.use_proto_names

tls

tls.client_certs[]

tls.client_certs[].cert

tls.client_certs[].cert_file

tls.client_certs[].key

tls.client_certs[].key_file

tls.client_certs[].password

tls.enable_renegotiation

tls.root_cas

tls.root_cas_file

tls.skip_cert_verify

url

Simple online edits

Contribution guide

`avro`

`avro.mapping`

`avro.preserve_logical_types`

`avro.raw_unions`

`avro.store_schema_metadata`

`avro.translate_kafka_connect_types`

`basic_auth`

`basic_auth.enabled`

`basic_auth.password`

`basic_auth.username`

`cache_duration`

`default_schema_id`

`jwt`

`jwt.claims`

`jwt.enabled`

`jwt.headers`

`jwt.private_key_file`

`jwt.signing_method`

`oauth`

`oauth.access_token`

`oauth.access_token_secret`

`oauth.consumer_key`

`oauth.consumer_secret`

`oauth.enabled`

`protobuf`

`protobuf.emit_default_values`

`protobuf.emit_unpopulated`

`protobuf.serialize_to_json`

`protobuf.use_enum_numbers`

`protobuf.use_proto_names`

`tls`

`tls.client_certs[]`

`tls.client_certs[].cert`

`tls.client_certs[].cert_file`

`tls.client_certs[].key`

`tls.client_certs[].key_file`

`tls.client_certs[].password`

`tls.enable_renegotiation`

`tls.root_cas`

`tls.root_cas_file`

`tls.skip_cert_verify`

`url`