Configuring Kafka serializers/deserializers in Java clients

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

Configuration for SerDes services

Configuration for SerDes lookup strategies

Configuration for Kafka converters

SchemaResolver interface

Apicurio 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.

DefaultSchemaResolver class

You can use the DefaultSchemaResolver to configure the following features:

Configuration for registry API access options

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

TLS security for the registry connection

When Apicurio Registry is served over HTTPS with certificates that the default Java trust store does not trust, you can configure a trust store for the SerDes HTTP client. You set the following properties directly as Kafka producer or consumer properties:

The following example configures a producer with a PKCS12 trust store:

Alternatively, provide one or more PEM certificate files without a password or type:

HTTP proxy for the registry connection

To reach Apicurio Registry through an HTTP proxy, set the apicurio.registry.proxy.host and apicurio.registry.proxy.port properties. For proxies that require authentication, also set apicurio.registry.proxy.username and apicurio.registry.proxy.password.

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 Apicurio 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:

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.

Configuration for registry lookup options

The DefaultSchemaResolver uses the following properties to configure how to look up artifacts in Apicurio Registry.

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 for deserializer fall-back options

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

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

You can configure application properties as Java system properties or include them in the Quarkus application.properties file. For more details, see the Quarkus documentation.

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

Kafka application configuration for serializers/deserializers

To use the SerDe classes that Apicurio 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:

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

Apicurio Registry provides the following Kafka client SerDes classes for Avro:

Avro serializer configuration

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

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:

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. Apicurio Registry provides the following implementations:

Apicurio 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. Apicurio Registry supports three different types:

The Apicurio Registry AvroDatumProvider class abstracts which type is used. The default is DefaultAvroDatumProvider.

You can set the following configuration options:

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 writer schema validation

Before writing a record, the Avro serializer verifies that the schema resolved from Apicurio Registry structurally matches the schema of the record being serialized. The schemas are compared by Avro parsing canonical form, so differences in properties that do not affect the binary encoding, such as connect.* metadata added by Kafka Connect, are tolerated. A structural mismatch, for example a stale latest artifact version when apicurio.registry.find-latest is enabled without apicurio.registry.auto-register, would otherwise silently write field values under the wrong fields. When a mismatch is detected, serialization fails with a descriptive error. Validation applies when the serialized object carries its own schema, as Avro generic and specific containers do, or when the resolved schema is a record and the configured datum provider can derive the object’s schema, as with the reflect provider. Plain Java collections without an attached schema are not validated.

You can disable this validation by setting the apicurio.registry.avro.validate-writer-schema property to false. The default is true.

Avro deserializer configuration

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

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

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, Apicurio Registry registers a new artifact for each nested record. For example, the following TradeKey schema includes a nested Exchange schema:

The nested Exchange schema is defined as follows:

When you use these schemas with Avro SerDes, Apicurio 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.

Custom logical type conversions for the Avro Kafka Connect converter

The io.apicurio.registry.utils.converter.AvroConverter Kafka Connect converter, provided by the apicurio-registry-utils-converter artifact, converts Kafka Connect logical types such as Decimal, Date, Time, and Timestamp to Avro logical types. You can register custom conversions for additional logical types, for example, Debezium micro-precision and nano-precision timestamp types, which have no built-in Avro mapping.

To add a custom conversion, implement the io.apicurio.registry.utils.converter.avro.ToAvroLogicalTypeConverter interface with the following methods:

Register the implementation through the Java ServiceLoader mechanism: package it in a JAR file with a META-INF/services/io.apicurio.registry.utils.converter.avro.ToAvroLogicalTypeConverter provider file that contains the implementation class name, and place the JAR file on the Kafka Connect worker classpath together with the converter. A custom conversion that declares the same Kafka Connect type name as a built-in conversion replaces the built-in conversion. Conversions apply in the Kafka Connect to Avro direction only.

Apicurio Registry provides the following Kafka client SerDes classes for JSON Schema:

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:

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:

JSON Schema deserializer configuration

You can configure the JSON Schema deserializer class as follows:

You must provide the location of Apicurio Registry so that the schema can be loaded. The other configuration is optional.

Custom ObjectMapper

The JSON Schema serializer and deserializer classes provide a setObjectMapper method that you can use to supply a custom Jackson ObjectMapper, for example, to register additional Jackson modules or custom serializers. You must call setObjectMapper before you call configure, and instantiate the SerDes class yourself instead of naming it in the producer or consumer properties:

When you do not set a custom mapper, the SerDes create a default ObjectMapper that ignores unknown properties during deserialization and omits null values during serialization.

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 Apicurio Registry.

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

The city.json schema is defined as follows:

In this example, a given citizen has a city. Apicurio 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.

Apicurio Registry provides the following Kafka client SerDes classes for Protobuf:

Protobuf serializer configuration

You can configure the Protobuf serializer class as follows:

Protobuf deserializer configuration

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

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

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.

Deserializer fallback parsing

By default, a schema resolution failure causes deserialization to fail. When you configure a specific return class, you can enable an opt-in fallback that parses the message payload directly with the generated class when schema lookup or parsing fails, for example, during a registry outage or when a transitive import is missing:

The fallback never applies when the deserializer returns DynamicMessage instances. When the fallback activates, the deserializer logs a warning that includes the original error.

Confluent interoperability

By default, the Protobuf SerDes use the Apicurio Registry wire format, which writes a type reference into the message payload. You can configure the following options to exchange Protobuf messages with clients that use the Confluent wire format, which writes a message-index list instead:

To produce messages that Confluent consumers can read, disable the type reference and enable message indexes on the serializer:

To consume messages in the Confluent format, mirror the settings on the deserializer, and set the return class so that the deserializer produces your generated message type:

Protobuf SerDes and artifact references

When you use a complex Protobuf message with an import statement, Apicurio Registry stores the imported Protobuf messages as separate artifacts. When Apicurio 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:

The mode.proto schema file is defined as follows:

In this example, Apicurio 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.