include::{mod-loc}shared/all-attributes.adoc[]

:_mod-docs-content-type: ASSEMBLY

[id="configuring-kafka-client-serdes_{context}"]
= Configuring Kafka serializers/deserializers in Java clients
//If the assembly covers a task, start the title with a verb in the gerund form, such as Creating or Configuring.

[role="_abstract"]
You can configure Kafka serializers/deserializers (SerDes) in your producer and consumer Java client applications:

* xref:registry-serdes-concepts-constants_{context}[]
* xref:registry-serdes-config-props_{context}[]
* xref:registry-serdes-types-serde_{context}[]
* xref:registry-serdes-types-avro_{context}[]
* xref:registry-serdes-types-json_{context}[]
* xref:registry-serdes-types-protobuf_{context}[]


.Prerequisites
* You have read {kafka-client-serdes}.

//INCLUDES


// Module included in the following assemblies:
//  assembly-using-kafka-client-serdes

:_mod-docs-content-type: REFERENCE

[id="registry-serdes-concepts-constants_{context}"]
== {registry} serializer/deserializer configuration in client applications

[role="_abstract"]
You can configure specific client serializer/deserializer (SerDes) services and schema lookup strategies directly in a client application by using the following example constants.

The following examples show commonly used SerDes constants and configuration options.

**Configuration for SerDes services**

[source,java,subs="+quotes,attributes"]
----
public class SerdeConfig {

   public static final String REGISTRY_URL = "apicurio.registry.url";
   public static final String ID_HANDLER = "apicurio.registry.id-handler";
----

* `REGISTRY_URL`: The required URL of {registry}.
* `ID_HANDLER`: Extends ID handling to support other ID formats and make them compatible with {registry} SerDes services.
For example, you can change it from the default ID format `Integer`, which is also compatible with the Confluent ID format, to `Long`.

**Configuration for SerDes lookup strategies**

[source,java,subs="+quotes,attributes"]
----
public class SerdeConfig {

   public static final String ARTIFACT_RESOLVER_STRATEGY = "apicurio.registry.artifact-resolver-strategy";
   public static final String SCHEMA_RESOLVER = "apicurio.registry.schema-resolver";
...
----

* `ARTIFACT_RESOLVER_STRATEGY`: Java class that implements the artifact resolver strategy and maps between the Kafka SerDes and artifact ID. Defaults to the topic ID strategy. Only the serializer class uses this option.
* `SCHEMA_RESOLVER`: Java class that implements the schema resolver. Defaults to `DefaultSchemaResolver`. The serializer and deserializer classes use this option.

**Configuration for Kafka converters**

[source,java,subs="+quotes,attributes"]
----
public class SerdeBasedConverter<S, T> extends SchemaResolverConfigurer<S, T> implements Converter, Closeable {

   public static final String REGISTRY_CONVERTER_SERIALIZER_PARAM = "apicurio.registry.converter.serializer";
   public static final String REGISTRY_CONVERTER_DESERIALIZER_PARAM = "apicurio.registry.converter.deserializer";
----

* `REGISTRY_CONVERTER_SERIALIZER_PARAM`: The required serializer to use with the {registry} Kafka converter.
* `REGISTRY_CONVERTER_DESERIALIZER_PARAM`: The required deserializer to use with the {registry} Kafka converter.

**Configuration for different schema types**

For details about how to configure SerDes for different schema technologies, see the following resources:

* xref:registry-serdes-types-avro_registry[]
* xref:registry-serdes-types-json_registry[]
* xref:registry-serdes-types-protobuf_registry[]

[role="_additional-resources"]
.Additional resources
* xref:registry-serdes-config-props_{context}[]
* {kafka-client-serdes}
* link:https://github.com/Apicurio/apicurio-registry/blob/main/utils/converter/src/main/java/io/apicurio/registry/utils/converter/SerdeBasedConverter.java[SerdeBasedConverter Java class]


// Metadata created by nebel
// ParentAssemblies: assemblies/getting-started/as_registry-reference.adoc

:_mod-docs-content-type: REFERENCE

[id="registry-serdes-config-props_{context}"]
== {registry} serializer/deserializer configuration properties

[role="_abstract"]
{registry} Kafka serializers/deserializers (SerDes) support the following Java configuration properties.

**SchemaResolver interface**

{registry} SerDes are based on the `SchemaResolver` interface, which abstracts access to the registry and applies the same lookup logic for the SerDes classes of all supported formats.

.Configuration property for SchemaResolver interface
[.table-expandable,width="100%",cols="5,5,5,3,5",options="header"]
|===
|Constant
|Property
|Description
|Type
|Default
|`SCHEMA_RESOLVER`
|`apicurio.registry.schema-resolver`
|Used by serializers and deserializers. Fully qualified Java class name that implements `SchemaResolver`.
|String
|`io.apicurio.registry.resolver.DefaultSchemaResolver`
|===

NOTE: Use the `DefaultSchemaResolver`, which provides useful features for most use cases.
For some advanced use cases, you might use a custom implementation of `SchemaResolver`.

**DefaultSchemaResolver class**

You can use the `DefaultSchemaResolver` to configure the following features:

* Access to the registry API
* How to look up artifacts in the registry
* How to write and read artifact information to and from Kafka
* Fall-back options for deserializers

**Configuration for registry API access options**

The `DefaultSchemaResolver` provides the following properties to configure access to the core registry API:

.Configuration properties for access to registry API
[.table-expandable,width="100%",cols="4,5,5,3,2",options="header"]
|===
|Constant
|Property
|Description
|Type
|Default
|`REGISTRY_URL`
|`apicurio.registry.url`
|Used by serializers and deserializers. URL to access the registry API.
|`String`
|None
|`AUTH_TOKEN_ENDPOINT`
|`apicurio.registry.auth.service.token.endpoint`
|Used by serializers and deserializers. URL of the token endpoint.
|`String`
|None
|`AUTH_CLIENT_ID`
|`apicurio.registry.auth.client.id`
|Used by serializers and deserializers. Client ID to access the authentication service. Required when accessing a secure registry by using the OAuth client credentials flow.
|`String`
|None
|`AUTH_CLIENT_SECRET`
|`apicurio.registry.auth.client.secret`
|Used by serializers and deserializers. Client secret to access the authentication service. Required when accessing a secure registry by using the OAuth client credentials flow.
|`String`
|None
|`AUTH_USERNAME`
|`apicurio.registry.auth.username`
|Used by serializers and deserializers. Username to access the registry. Required when accessing a secure registry by using HTTP basic authentication.
|`String`
|None
|`AUTH_PASSWORD`
|`apicurio.registry.auth.password`
|Used by serializers and deserializers. Password to access the registry. Required when accessing a secure registry by using HTTP basic authentication.
|`String`
|None
|===

**OpenTelemetry distributed tracing**

The SerDes libraries support optional OpenTelemetry trace context propagation. When enabled,
W3C trace context headers (`traceparent` and `tracestate`) are injected into all HTTP requests
to {registry}, allowing schema resolution and registration calls to appear as spans in your
distributed traces alongside your Kafka producer and consumer operations.

To enable this feature, set the following configuration property:

[source,properties]
----
apicurio.registry.otel.enabled=true
----

The `opentelemetry-api` library must be on the classpath. When you initialize an OpenTelemetry SDK
in your application (for example, by using the OpenTelemetry Java agent or the Quarkus OpenTelemetry
extension), the SerDes HTTP calls participate in the active trace. If no OpenTelemetry SDK is
configured, the behavior is a no-op and the SerDes add no trace headers.

For more details about SDK-level configuration, see xref:registry-sdk-config_{context}[].

**Configuration for registry lookup options**

The `DefaultSchemaResolver` uses the following properties to configure how to look up artifacts in {registry}.

.Configuration properties for registry artifact lookup
[.table-expandable,width="100%",cols="5,5,5,3,4",options="header"]
|===
|Constant
|Property
|Description
|Type
|Default
|`ARTIFACT_RESOLVER_STRATEGY`
|`apicurio.registry.artifact-resolver-strategy`
|Used by serializers only. Fully qualified Java class name that implements `ArtifactReferenceResolverStrategy` and maps each Kafka message to an `ArtifactReference` (`groupId`, `artifactId`, and version). For example, the default strategy uses the topic name as the schema `artifactId`.
|`String`
|`io.apicurio.registry.serde.strategy.TopicIdStrategy`
|`EXPLICIT_ARTIFACT_GROUP_ID`
|`apicurio.registry.artifact.group-id`
|Used by serializers only. Sets the `groupId` used for querying or creating an artifact. Overrides the `groupId` returned by the `ArtifactResolverStrategy`.
|`String`
|None
|`EXPLICIT_ARTIFACT_ID`
|`apicurio.registry.artifact.artifact-id`
|Used by serializers only. Sets the `artifactId` used for querying or creating an artifact. Overrides the `artifactId` returned by the `ArtifactResolverStrategy`.
|`String`
|None
|`EXPLICIT_ARTIFACT_VERSION`
|`apicurio.registry.artifact.version`
|Used by serializers only. Sets the artifact version used for querying or creating an artifact. Overrides the version returned by the `ArtifactResolverStrategy`.
|`String`
|None
|`FIND_LATEST_ARTIFACT`
|`apicurio.registry.find-latest`
|Used by serializers only. Specifies whether the serializer tries to find the latest artifact in the registry for the corresponding group ID and artifact ID.
|`boolean`
|`false`
|`AUTO_REGISTER_ARTIFACT`
|`apicurio.registry.auto-register`
|Used by serializers only. Specifies whether the serializer tries to create an artifact in the registry. The JSON Schema serializer does not support this feature.
|`boolean, boolean String`
|`false`
|`DEREFERENCE_SCHEMA`
|`apicurio.registry.dereference-schema`
|Used by serializers and deserializers. Configures the SerDes to dereference the schema. After the schema is registered, this property instructs the SerDes to request the dereferenced schema from the server. It also instructs the serializer to dereference the schema before registering it in {registry}, but this is only supported for Avro.
|`boolean`
|`false`
|`AUTO_REGISTER_ARTIFACT_IF_EXISTS`
|`apicurio.registry.auto-register.if-exists`
|Used by serializers only. Configures the behavior of the client when a conflict occurs while creating an artifact because the artifact already exists. Available values are `FAIL`, `UPDATE`, `RETURN`, or `RETURN_OR_UPDATE`.
|`String`
|`RETURN_OR_UPDATE`
|`CHECK_PERIOD_MS`
|`apicurio.registry.check-period-ms`
|Used by serializers and deserializers. Specifies how long to cache artifacts before auto-eviction (milliseconds). If set to zero, artifacts are fetched every time.
|`java.time.Duration, non-negative Number, or integer String`
|`30000`
|`RETRY_BACKOFF_MS`
|`apicurio.registry.retry-backoff-ms`
|Used by serializers and deserializers. If the SerDes cannot retrieve a schema from the registry, they can retry a number of times. This configuration option controls the delay between the retry attempts (milliseconds).
|`java.time.Duration, non-negative Number, or integer String`
|`300`
|`RETRY_COUNT`
|`apicurio.registry.retry-count`
|Used by serializers and deserializers. If the SerDes cannot retrieve a schema from the registry, they can retry a number of times. This configuration option controls the number of retry attempts.
|`non-negative Number, or integer String`
|`3`
|`FAULT_TOLERANT_REFRESH`
|`apicurio.registry.fault-tolerant-refresh`
|Used by serializers and deserializers. If `true`, the SerDes log exceptions instead of throwing them when an error occurs while refreshing a schema in the cache. This is useful for production situations where a stale schema is better than completely failing schema resolution. This does not affect retry attempts, because the SerDes attempt retries before considering this flag.
|`boolean`
|`false`
|`BACKGROUND_REFRESH_ENABLED`
|`apicurio.registry.background-refresh-enabled`
|Used by serializers and deserializers. If `true`, enables background refresh of expired cache entries following the "stale-while-revalidate" pattern. When an entry expires and a stale value exists, the cache returns the stale value immediately (non-blocking) and triggers an asynchronous refresh in the background. This prevents latency spikes in high-concurrency scenarios where multiple threads would otherwise block waiting for cache refresh. If no stale value exists (first fetch), the cache falls back to synchronous refresh behavior. When a background refresh fails, the cache logs errors and continues to serve stale values (similar to `FAULT_TOLERANT_REFRESH` behavior). This option complements `FAULT_TOLERANT_REFRESH`, which handles refresh failures rather than refresh latency.
|`boolean`
|`false`
|`BACKGROUND_REFRESH_EXECUTOR_THREADS`
|`apicurio.registry.background-refresh.executor-threads`
|Used by serializers and deserializers when `BACKGROUND_REFRESH_ENABLED` is enabled. Specifies the number of threads in the background refresh executor pool. The executor uses a fixed thread pool with daemon threads to prevent blocking application shutdown. A value of 2-4 threads is typically sufficient for most workloads.
|`non-negative Number, or integer String`
|`2`
|`BACKGROUND_REFRESH_TIMEOUT_MS`
|`apicurio.registry.background-refresh.timeout-ms`
|Used by serializers and deserializers when `BACKGROUND_REFRESH_ENABLED` is enabled. Specifies the maximum time (milliseconds) to wait for a background refresh operation to complete before timing out. If a background refresh exceeds this timeout, it is interrupted and the cache continues to serve the stale value until the next refresh attempt.
|`java.time.Duration, non-negative Number, or integer String`
|`30000`
|`USE_ID`
|`apicurio.registry.use-id`
|Used by serializers and deserializers. Configures the SerDes to use the specified `IdOption` as the identifier for artifacts. Options are `globalId` and `contentId`. Instructs the serializer to write the specified ID to Kafka, and instructs the deserializer to use this ID to find the schema.
|`String`
|`contentId`
|`CANONICALIZE`
|`apicurio.registry.canonicalize`
|Used by serializers only. Specifies whether the serializer canonicalizes schema content when looking up or creating an artifact version in the registry by content. When set to `true`, semantically equivalent schemas with different formatting match. When `false`, schema content is compared as-is. NOTE: In versions before 3.1.0, canonicalization was always attempted as a fallback. Starting with 3.1.0, you must explicitly set this property to `true` if you want canonicalized lookups.
|`boolean`
|`false`
|===

**Configuration to read and write registry artifacts in Kafka**

The `DefaultSchemaResolver` uses the following properties to configure how artifact information is written to and read from Kafka.

.Configuration properties to read/write artifact information in Kafka
[.table-expandable,width="100%",cols="5,5,5,3,5",options="header"]
|===
|Constant
|Property
|Description
|Type
|Default
|`ENABLE_HEADERS`
|`apicurio.registry.headers.enabled`
|Used by serializers and deserializers. Configures the SerDes to read and write the artifact identifier in Kafka message headers instead of in the message payload.
|`boolean`
|`false`
|`HEADERS_HANDLER`
|`apicurio.registry.headers.handler`
|Used by serializers and deserializers. Fully qualified Java class name that implements `HeadersHandler` and writes and reads the artifact identifier to and from the Kafka message headers.
|`String`
|`io.apicurio.registry.serde.headers.DefaultHeadersHandler`
|`ID_HANDLER`
|`apicurio.registry.id-handler`
|Used by serializers and deserializers. Fully qualified Java class name of a class that implements `IdHandler` and writes and reads the artifact identifier to and from the message payload. Defaults to a 4-byte format that includes the `contentId` in the message payload.
|`String`
|`io.apicurio.registry.serde.Default4ByteIdHandler`
|===

**Configuration for deserializer fall-back options**

The `DefaultSchemaResolver` uses the following property to configure a fall-back provider for all deserializers.

.Configuration property for deserializer fall-back provider
[.table-expandable,width="100%",cols="5,5,5,3,5",options="header"]
|===
|Constant
|Property
|Description
|Type
|Default
|`FALLBACK_ARTIFACT_PROVIDER`
|`apicurio.registry.fallback.provider`
|Used by deserializers only. Sets a custom implementation of `FallbackArtifactProvider` for resolving the artifact used for deserialization. `FallbackArtifactProvider` configures a fallback artifact to fetch from the registry if the lookup fails.
|`String`
|`io.apicurio.registry.serde.fallback.DefaultFallbackArtifactProvider`
|===

The `DefaultFallbackArtifactProvider` uses the following properties to configure deserializer fall-back options:

.Configuration properties for deserializer fall-back options
[.table-expandable,width="100%",cols="5,5,5,3,2",options="header"]
|===
|Constant
|Property
|Description
|Type
|Default
|`FALLBACK_ARTIFACT_ID`
|`apicurio.registry.fallback.artifact-id`
|Used by deserializers only. Sets the `artifactId` used as fallback for resolving the artifact used for deserialization.
|`String`
|None
|`FALLBACK_ARTIFACT_GROUP_ID`
|`apicurio.registry.fallback.group-id`
|Used by deserializers only. Sets the `groupId` used as fallback for resolving the group used for deserialization.
|`String`
|None
|`FALLBACK_ARTIFACT_VERSION`
|`apicurio.registry.fallback.version`
|Used by deserializers only. Sets the version used as fallback for resolving the artifact used for deserialization.
|`String`
|None
|===

You can configure application properties as Java system properties or include them in the Quarkus
`application.properties` file.
For more details, see the https://quarkus.io/guides/config#overriding-properties-at-runtime[Quarkus documentation].

[role="_additional-resources"]
.Additional resources
* link:https://github.com/Apicurio/apicurio-registry/blob/main/serdes/generic/serde-common/src/main/java/io/apicurio/registry/serde/config/SerdeConfig.java[SerdeConfig Java class]


// Module included in the following assemblies:
//  assembly-using-kafka-client-serdes

:_mod-docs-content-type: REFERENCE

[id="registry-serdes-types-serde_{context}"]
== Client serializer/deserializer types

[role="_abstract"]
When you use schemas in your Kafka client applications, you must choose which specific schema type to use, depending on your use case. {registry} provides SerDe Java classes for Apache Avro, JSON Schema, and Google Protobuf.

You can also use Kafka to implement custom serializer and deserializer classes, and use {registry} functions through the {registry} REST Java client.

**Kafka application configuration for serializers/deserializers**

To use the SerDe classes that {registry} provides in your Kafka application, set the correct configuration properties. The following Avro examples show how to configure a serializer in a Kafka producer application and how to configure a deserializer in a Kafka consumer application.

The following example shows a serializer configuration in a Kafka producer:

[source,java,subs="+quotes,attributes"]
----
// Create the Kafka producer
private static Producer<Object, Object> createKafkaProducer() {
    Properties props = new Properties();

    // Configure standard Kafka settings
    props.putIfAbsent(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, SERVERS);
    props.putIfAbsent(ProducerConfig.CLIENT_ID_CONFIG, "Producer-" + TOPIC_NAME);
    props.putIfAbsent(ProducerConfig.ACKS_CONFIG, "all");

    // Use {registry}-provided Kafka serializer for Avro
    props.putIfAbsent(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    props.putIfAbsent(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, AvroKafkaSerializer.class.getName());

    // Configure the {registry} location
    props.putIfAbsent(SerdeConfig.REGISTRY_URL, REGISTRY_URL);

    // Register the schema artifact if not found in the registry.
    props.putIfAbsent(SerdeConfig.AUTO_REGISTER_ARTIFACT, Boolean.TRUE);

    // Create the Kafka producer
    Producer<Object, Object> producer = new KafkaProducer<>(props);
    return producer;
}
----

The following example shows a deserializer configuration in a Kafka consumer:

[source,java,subs="+quotes,attributes"]
----
// Create the Kafka consumer
private static KafkaConsumer<Long, GenericRecord> createKafkaConsumer() {
    Properties props = new Properties();

    // Configure standard Kafka settings
    props.putIfAbsent(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, SERVERS);
    props.putIfAbsent(ConsumerConfig.GROUP_ID_CONFIG, "Consumer-" + TOPIC_NAME);
    props.putIfAbsent(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
    props.putIfAbsent(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
    props.putIfAbsent(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

    // Use {registry}-provided Kafka deserializer for Avro
    props.putIfAbsent(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    props.putIfAbsent(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, AvroKafkaDeserializer.class.getName());

    // Configure the {registry} location
    props.putIfAbsent(SerdeConfig.REGISTRY_URL, REGISTRY_URL);

    // No other configuration needed because the schema globalId the deserializer uses is sent
    // in the payload. The deserializer extracts the globalId and uses it to look up the schema
    // from the registry.

    // Create the Kafka consumer
    KafkaConsumer<Long, GenericRecord> consumer = new KafkaConsumer<>(props);
    return consumer;
}
----

[role="_additional-resources"]
.Additional resources
* link:https://github.com/Apicurio/apicurio-registry/tree/main/examples[Simple Avro example]


// Module included in the following assemblies:
//  assembly-using-kafka-client-serdes

:_mod-docs-content-type: REFERENCE

[id="registry-serdes-types-avro_{context}"]
== Avro SerDes configuration

[role="_abstract"]
You can configure the Kafka client serializer and deserializer (SerDes) classes for Apache Avro by using the configuration options in this section.

{registry} provides the following Kafka client SerDes classes for Avro:

* `io.apicurio.registry.serde.avro.AvroKafkaSerializer`
* `io.apicurio.registry.serde.avro.AvroKafkaDeserializer`

**Avro serializer configuration**

You can configure the Avro serializer class with the following options:

* {registry} URL
* Artifact resolver strategy
* ID location
* ID encoding
* Avro datum provider
* Avro encoding

**ID location**

The serializer passes the unique ID of the schema as part of the Kafka message so that consumers can use the correct schema for deserialization. The ID can be in the message payload or in the message headers. The default location is the message payload. To send the ID in the message headers, set the following configuration property:
----
props.putIfAbsent(SerdeConfig.ENABLE_HEADERS, "true")
----
The property name is `apicurio.registry.headers.enabled`.

**ID encoding**

You can customize how the SerDes encode the schema ID when they pass it in the Kafka message body. Set the `apicurio.registry.id-handler` configuration property to a class that implements the `io.apicurio.registry.serde.IdHandler` interface. {registry} provides the following implementations:

* `io.apicurio.registry.serde.Default4ByteIdHandler`: Stores the ID as a 4-byte integer
* `io.apicurio.registry.serde.Legacy8ByteIdHandler`: Stores the ID as an 8-byte long

{registry} represents the schema ID as a long, but for legacy reasons, or for compatibility with other registries or SerDe classes, you might want to use 4 bytes when sending the ID.

**Avro datum provider**

Avro provides different datum writers and readers to write and read data. {registry} supports three different types:

* Generic
* Specific
* Reflect

The {registry} `AvroDatumProvider` class abstracts which type is used. The default is `DefaultAvroDatumProvider`.

You can set the following configuration options:

* `apicurio.registry.avro-datum-provider`: Specifies a fully qualified Java class name of the `AvroDatumProvider` implementation, for example `io.apicurio.registry.serde.avro.ReflectAvroDatumProvider`
* `apicurio.registry.use-specific-avro-reader`: Specifies whether to use a specific type when you use `DefaultAvroDatumProvider`. Set to `true` to enable this option.

**Avro encoding**

When you use Avro to serialize data, you can use the Avro binary encoding format to ensure that the data is encoded in as efficient a format as possible. Avro also supports encoding the data as JSON, which you can use to inspect the payload of each message, for example, for logging or debugging.

You can set the Avro encoding by configuring the `apicurio.registry.avro.encoding` property with a value of `JSON` or `BINARY`. The default is `BINARY`.

**Avro deserializer configuration**

You must configure the Avro deserializer class to match the following configuration settings of the serializer:

* {registry} URL
* ID encoding
* Avro datum provider
* Avro encoding

See the serializer section for these configuration options. The property names and values are the same.

[NOTE]
====
The following options are not required when configuring the deserializer:

* Artifact resolver strategy
* ID location
====

The deserializer class can determine the values for these options from the message. The strategy is not required because the serializer is responsible for sending the ID as part of the message.

The deserializer determines the ID location by checking for the magic byte at the start of the message payload. If that byte is found, the deserializer reads the ID from the message payload by using the configured handler. If the magic byte is not found, the deserializer reads the ID from the message headers.

**Avro SerDes and artifact references**

When you work with Avro messages and a schema with nested records, {registry} registers a new artifact for each nested record. For example, the following `TradeKey` schema includes a nested `Exchange` schema:

[source,json]
----
{
  "namespace": "com.kubetrade.schema.trade",
  "type": "record",
  "name": "TradeKey",
  "fields": [
    {
      "name": "exchange",
      "type": "com.kubetrade.schema.common.Exchange"
    },
    {
      "name": "key",
      "type": "string"
    }
  ]
}
----

The nested `Exchange` schema is defined as follows:

[source,json]
----
{
  "namespace": "com.kubetrade.schema.common",
  "type": "enum",
  "name": "Exchange",
  "symbols" : ["GEMINI"]
}
----

When you use these schemas with Avro SerDes, {registry} creates two artifacts, one for the `TradeKey` schema and one for the `Exchange` schema. Whenever a message that uses the `TradeKey` schema is serialized or deserialized, both schemas are retrieved, allowing you to split your definitions into different files.

[role="_additional-resources"]
.Additional resources
* link:https://github.com/Apicurio/apicurio-registry/blob/main/serdes/avro-serde/src/main/java/io/apicurio/registry/serde/avro/AvroSerdeConfig.java[AvroSerdeConfig Java class]
* link:https://github.com/Apicurio/apicurio-registry/tree/main/examples[Simple Avro example]
* link:https://github.com/Apicurio/apicurio-registry/tree/main/examples[SerDes with references example]


// Module included in the following assemblies:
//  assembly-using-kafka-client-serdes

:_mod-docs-content-type: REFERENCE

[id="registry-serdes-types-json_{context}"]
== JSON Schema SerDes configuration

[role="_abstract"]
You can configure the Kafka client serializer and deserializer (SerDes) classes for JSON Schema by using the configuration options in this section.

{registry} provides the following Kafka client SerDes classes for JSON Schema:

* `io.apicurio.registry.serde.jsonschema.JsonSchemaKafkaSerializer`
* `io.apicurio.registry.serde.jsonschema.JsonSchemaKafkaDeserializer`

Unlike Apache Avro, JSON Schema is not a serialization technology, but is instead a validation technology. As a result, configuration options for JSON Schema are different. For example, no encoding option exists, because data is always encoded as JSON.

**JSON Schema serializer configuration**

You can configure the JSON Schema serializer class as follows:

* {registry} URL
* Artifact resolver strategy
* Schema validation

The only non-standard configuration property is JSON Schema validation, which is enabled by default. You can disable this by setting
`apicurio.registry.serde.validation-enabled` to `"false"`. For example:
----
props.putIfAbsent(SerdeConfig.VALIDATION_ENABLED, Boolean.FALSE)
----

**JSON Schema deserializer configuration**

You can configure the JSON Schema deserializer class as follows:

* {registry} URL
* Schema validation
* Class for deserializing data

You must provide the location of {registry} so that the schema can be loaded. The other configuration is optional.

NOTE: Deserializer validation only works if the serializer passes the global ID in the Kafka message, which only happens when you enable validation in the serializer.

**JSON Schema SerDes and artifact references**

The JSON Schema SerDes cannot discover the schema from the message payload, so you must register the schema artifact beforehand, and this also applies to artifact references.

If the `$ref` value is a URL, the SerDes try to resolve the referenced schema by using that URL. Validation then works as usual: the SerDes validate the data against the main schema and validate the nested value against the nested schema. The SerDes also support referencing artifacts in {registry}.

For example, the following `citizen.json` schema references the `city.json` schema:

[source,json]
----
{
 "$id": "https://example.com/citizen.schema.json",
 "$schema": "http://json-schema.org/draft-07/schema#",
 "title": "Citizen",
 "type": "object",
 "properties": {
   "firstName": {
     "type": "string",
     "description": "The citizen's first name."
   },
   "lastName": {
     "type": "string",
     "description": "The citizen's last name."
   },
   "age": {
     "description": "Age in years which must be equal to or greater than zero.",
     "type": "integer",
     "minimum": 0
   },
   "city": {
     "$ref": "city.json"
   }
 }
}
----

The `city.json` schema is defined as follows:

[source,json]
----
{
 "$id": "https://example.com/city.schema.json",
 "$schema": "http://json-schema.org/draft-07/schema#",
 "title": "City",
 "type": "object",
 "properties": {
   "name": {
     "type": "string",
     "description": "The city's name."
   },
   "zipCode": {
     "type": "integer",
     "description": "The zip code.",
     "minimum": 0
   }
 }
}
----

In this example, a given citizen has a city. {registry} creates a citizen artifact with a reference to the city artifact by using the name `city.json`. In the SerDes, when the citizen schema is fetched, the city schema is also fetched because it is referenced from the citizen schema. When you serialize or deserialize data, the SerDes use the reference name to resolve the nested schema, allowing validation against the citizen schema and the nested city schema.

[role="_additional-resources"]
.Additional resources
* link:https://github.com/Apicurio/apicurio-registry/blob/main/serdes/jsonschema-serde/src/main/java/io/apicurio/registry/serde/jsonschema/JsonSchemaKafkaDeserializerConfig.java[JsonSchemaKafkaDeserializerConfig Java class]
* link:https://github.com/Apicurio/apicurio-registry/tree/main/examples[Simple JSON Schema example]
* link:https://github.com/Apicurio/apicurio-registry/tree/main/examples[SerDes with references example]


// Module included in the following assemblies:
//  assembly-using-kafka-client-serdes

:_mod-docs-content-type: REFERENCE

[id="registry-serdes-types-protobuf_{context}"]
== Protobuf SerDes configuration

[role="_abstract"]
You can configure the Kafka client serializer and deserializer (SerDes) classes for Google Protobuf by using the configuration options in this section.

{registry} provides the following Kafka client SerDes classes for Protobuf:

* `io.apicurio.registry.serde.protobuf.ProtobufKafkaSerializer`
* `io.apicurio.registry.serde.protobuf.ProtobufKafkaDeserializer`

**Protobuf serializer configuration**

You can configure the Protobuf serializer class as follows:

* {registry} URL
* Artifact resolver strategy
* ID location
* ID encoding
* Schema validation

For details about these configuration options, see the following sections:

* xref:registry-serdes-concepts-constants_registry[]
* xref:registry-serdes-types-avro_registry[]

**Protobuf deserializer configuration**

You must configure the Protobuf deserializer class to match the following configuration settings in the serializer:

* {registry} URL
* ID encoding

The configuration property names and values are the same as for the serializer.

[NOTE]
====
The following options are not required when configuring the deserializer:

* Artifact resolver strategy
* ID location
====

The deserializer class can determine the values for these options from the message. The strategy is not required because the serializer is responsible for sending the ID as part of the message.

The deserializer determines the ID location by checking for the magic byte at the start of the message payload. If that byte is found, the deserializer reads the ID from the message payload by using the configured handler. If the magic byte is not found, the deserializer reads the ID from the message headers.

NOTE: The Protobuf deserializer does not deserialize to your exact Protobuf Message implementation, but rather to a `DynamicMessage` instance. No appropriate API exists to do otherwise.

**Protobuf SerDes and artifact references**

When you use a complex Protobuf message with an `import` statement, {registry} stores the imported Protobuf messages as separate artifacts. When {registry} gets the main schema to check a Protobuf message, it also retrieves the referenced schemas so that the full message schema can be checked and serialized.

For example, the following `table_info.proto` schema file includes the imported `mode.proto` schema file:

[source,bash]
----
syntax = "proto3";
package sample;
option java_package = "io.api.sample";
option java_multiple_files = true;

import "sample/mode.proto";

message TableInfo {

 int32 winIndex = 1;
 Mode mode = 2;
 int32 min = 3;
 int32 max = 4;
 string id = 5;
 string dataAdapter = 6;
 string schema = 7;
 string selector = 8;
 string subscription_id = 9;
}
----

The `mode.proto` schema file is defined as follows:

[source,bash]
----
syntax = "proto3";
package sample;
option java_package = "io.api.sample";
option java_multiple_files = true;

enum Mode {

MODE_UNKNOWN = 0;
RAW = 1;
MERGE = 2;
DISTINCT = 3;
COMMAND = 4;
}
----

In this example, {registry} stores two Protobuf artifacts, one for `TableInfo` and one for `Mode`. However, because `Mode` is part of `TableInfo`, whenever `TableInfo` is fetched to check a message in the SerDes, `Mode` is also returned as an artifact referenced by `TableInfo`.

[role="_additional-resources"]
.Additional resources
* link:https://github.com/Apicurio/apicurio-registry/tree/main/examples[Protobuf Bean and Protobuf Find Latest examples]
* link:https://github.com/Apicurio/apicurio-registry/tree/main/examples[SerDes with references example]
