# schema_registry_decode

> For the complete documentation index, see [llms.txt](https://docs.redpanda.com/llms.txt). Component-specific: [cloud-data-platform-full.txt](https://docs.redpanda.com/cloud-data-platform-full.txt)

---
title: schema_registry_decode
latest-operator-version: v26.1.4
latest-console-tag: v3.7.3
latest-connect-version: 4.93.0
latest-redpanda-tag: v26.1.9
docname: connect/components/processors/schema_registry_decode
page-component-name: cloud-data-platform
page-version: master
page-component-version: master
page-component-title: Cloud
page-relative-src-path: connect/components/processors/schema_registry_decode.adoc
page-edit-url: https://github.com/redpanda-data/cloud-docs/edit/main/modules/develop/pages/connect/components/processors/schema_registry_decode.adoc
page-git-created-date: "2024-09-09"
page-git-modified-date: "2026-05-26"
---

<!-- Source: https://docs.redpanda.com/cloud-data-platform/develop/connect/components/processors/schema_registry_decode.md -->

**Available in:** Cloud, [Self-Managed](https://docs.redpanda.com/connect/components/processors/schema_registry_decode/%20%22View%20the%20Self-Managed%20version%20of%20this%20component%22)

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

This processor uses the [Franz Kafka Schema Registry client](https://github.com/twmb/franz-go/tree/master/pkg/sr).

#### Common

```yml
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)
```

#### Advanced

```yml
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](https://docs.confluent.io/platform/current/schema-registry/index.html) 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](https://docs.redpanda.com/cloud-data-platform/develop/connect/configuration/error_handling/).

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

## [](#avro-json-format)Avro JSON format

By default, this processor expects documents formatted as [Avro JSON](https://avro.apache.org/docs/current/specification/) 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](https://pkg.go.dev/github.com/linkedin/goavro/v2#NewCodecForStandardJSONFull) by setting the field [`avro.raw_unions`](#avro-raw_unions) to `true`.

## [](#protobuf-format)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](https://developers.google.com/protocol-buffers/docs/proto3#json).

## [](#metadata)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)Fields

### [](#avro)`avro`

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

**Type**: `object`

### [](#avro-mapping)`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`

```yaml
# 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)`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)`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](https://avro.apache.org/docs/current/specification/).

If set to `false`, Avro messages are decoded as [Avro JSON](https://pkg.go.dev/github.com/linkedin/goavro/v2#NewCodec).

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](https://pkg.go.dev/github.com/linkedin/goavro/v2#NewCodecForStandardJSONFull).

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](https://github.com/linkedin/goavro/blob/5ec5a5ee7ec82e16e6e2b438d610e1cab2588393/union.go#L224-L249) and the [underlying library used for Avro serialization](https://github.com/linkedin/goavro).

**Type**: `bool`

### [](#avro-store_schema_metadata)`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)`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:

| 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)`basic_auth`

Allows you to specify basic authentication.

**Type**: `object`

### [](#basic_auth-enabled)`basic_auth.enabled`

Whether to use basic authentication in requests.

**Type**: `bool`

**Default**: `false`

### [](#basic_auth-password)`basic_auth.password`

A password to authenticate with.

> ⚠️ **CAUTION**
>
> This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see [Manage Secrets](https://docs.redpanda.com/cloud-data-platform/develop/connect/configuration/secret-management/) before adding it to your configuration.

**Type**: `string`

**Default**: `""`

### [](#basic_auth-username)`basic_auth.username`

A username to authenticate as.

**Type**: `string`

**Default**: `""`

### [](#cache_duration)`cache_duration`

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

**Type**: `string`

**Default**: `10m`

```yaml
# Examples:
cache_duration: 1h

# ---

cache_duration: 5m
```

### [](#default_schema_id)`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.

> 💡 **TIP**
>
> You can also use the [`with_schema_registry_header`](https://docs.redpanda.com/cloud-data-platform/develop/connect/guides/bloblang/functions/#with_schema_registry_header) bloblang function to add a schema ID to messages.

**Type**: `int`

### [](#jwt)`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)`jwt.claims`

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

**Type**: `object`

**Default**: `{}`

### [](#jwt-enabled)`jwt.enabled`

Whether to use JWT authentication in requests.

**Type**: `bool`

**Default**: `false`

### [](#jwt-headers)`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)`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)`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)`oauth`

Configure OAuth version 1.0 authentication for secure API access.

**Type**: `object`

### [](#oauth-access_token)`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)`oauth.access_token_secret`

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

> ⚠️ **CAUTION**
>
> This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see [Manage Secrets](https://docs.redpanda.com/cloud-data-platform/develop/connect/configuration/secret-management/) before adding it to your configuration.

**Type**: `string`

**Default**: `""`

### [](#oauth-consumer_key)`oauth.consumer_key`

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

**Type**: `string`

**Default**: `""`

### [](#oauth-consumer_secret)`oauth.consumer_secret`

A secret used to establish ownership of the consumer key.

> ⚠️ **CAUTION**
>
> This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see [Manage Secrets](https://docs.redpanda.com/cloud-data-platform/develop/connect/configuration/secret-management/) before adding it to your configuration.

**Type**: `string`

**Default**: `""`

### [](#oauth-enabled)`oauth.enabled`

Whether to use OAuth version 1 in requests.

**Type**: `bool`

**Default**: `false`

### [](#protobuf)`protobuf`

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

**Type**: `object`

### [](#protobuf-emit_default_values)`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)`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)`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)`protobuf.use_enum_numbers`

Emits enum values as numbers.

**Type**: `bool`

**Default**: `false`

### [](#protobuf-use_proto_names)`protobuf.use_proto_names`

Use proto field name instead of lowerCamelCase name.

**Type**: `bool`

**Default**: `false`

### [](#tls)`tls`

Custom TLS settings can be used to override system defaults.

**Type**: `object`

### [](#tls-client_certs)`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**: `[]`

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

# ---

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

### [](#tls-client_certs-cert)`tls.client_certs[].cert`

A plain text certificate to use.

**Type**: `string`

**Default**: `""`

### [](#tls-client_certs-cert_file)`tls.client_certs[].cert_file`

The path of a certificate to use.

**Type**: `string`

**Default**: `""`

### [](#tls-client_certs-key)`tls.client_certs[].key`

A plain text certificate key to use.

> ⚠️ **CAUTION**
>
> This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see [Manage Secrets](https://docs.redpanda.com/cloud-data-platform/develop/connect/configuration/secret-management/) before adding it to your configuration.

**Type**: `string`

**Default**: `""`

### [](#tls-client_certs-key_file)`tls.client_certs[].key_file`

The path of a certificate key to use.

**Type**: `string`

**Default**: `""`

### [](#tls-client_certs-password)`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.

> ⚠️ **CAUTION**
>
> This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see [Manage Secrets](https://docs.redpanda.com/cloud-data-platform/develop/connect/configuration/secret-management/) before adding it to your configuration.

**Type**: `string`

**Default**: `""`

```yaml
# Examples:
password: foo

# ---

password: ${KEY_PASSWORD}
```

### [](#tls-enable_renegotiation)`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)`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.

> ⚠️ **CAUTION**
>
> This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see [Manage Secrets](https://docs.redpanda.com/cloud-data-platform/develop/connect/configuration/secret-management/) before adding it to your configuration.

**Type**: `string`

**Default**: `""`

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

### [](#tls-root_cas_file)`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**: `""`

```yaml
# Examples:
root_cas_file: ./root_cas.pem
```

### [](#tls-skip_cert_verify)`tls.skip_cert_verify`

Whether to skip server side certificate verification.

**Type**: `bool`

**Default**: `false`

### [](#url)`url`

The base URL of the schema registry service.

**Type**: `string`