Schema Registry

Dekaf integrates with Confluent Schema Registry for schema management and evolution with Avro and Protobuf serialization.

Installation

# Core Schema Registry support
dotnet add package Dekaf.SchemaRegistry

# For Avro serialization
dotnet add package Dekaf.SchemaRegistry.Avro

# For Protobuf serialization
dotnet add package Dekaf.SchemaRegistry.Protobuf

# Optional JSON Schema payload validation
dotnet add package Dekaf.SchemaRegistry.Json

Dekaf.SchemaRegistry keeps JSON validation disabled by default and does not depend on a JSON Schema engine. Install Dekaf.SchemaRegistry.Json only in applications that enable validation.

Association-backed subject names

SubjectNameStrategy.AssociatedName resolves the key or value subject from a Schema Registry topic association. This is opt-in and leaves the default topic-name behavior unchanged. Association lookups happen during asynchronous serializer/deserializer preparation; warmed serialization does not perform network calls, take locks, allocate, or enter an async state machine.

using Avro;
using Avro.Generic;
using Dekaf.SchemaRegistry;
using Dekaf.SchemaRegistry.Avro;

var registry = new SchemaRegistryClient(new SchemaRegistryConfig
{
    Url = "https://schema-registry.example.com"
});

var orderSchema = (RecordSchema)Avro.Schema.Parse("""
    {
      "type": "record",
      "name": "Order",
      "namespace": "example.orders",
      "fields": [{ "name": "id", "type": "string" }]
    }
    """);
var order = new GenericRecord(orderSchema);
order.Add("id", "order-123");

var serializer = new AvroSchemaRegistrySerializer<GenericRecord>(
    registry,
    new AvroSerializerConfig
    {
        SubjectNameStrategy = SubjectNameStrategy.AssociatedName,
        AutoRegisterSchemas = false,
        UseLatestVersion = true
    });

await serializer.PrepareAsync("orders", order);

The association resourceName must equal the Kafka topic. Use association type key for key serdes and value for value serdes. By default Dekaf queries the wildcard resource namespace -, which matches associations from any Kafka cluster. For cluster-specific governance, share an explicit strategy instance:

var associatedNames = new AssociatedNameStrategy(
    registry,
    new AssociatedNameStrategyOptions
    {
        KafkaClusterId = "lkc-12345",
        FallbackStrategy = AssociatedNameFallbackStrategy.None
    });

var config = new AvroSerializerConfig
{
    AsyncSubjectNameStrategy = associatedNames,
    AutoRegisterSchemas = false
};

CustomSubjectNameStrategy has highest precedence. When it is null, AsyncSubjectNameStrategy takes precedence over the enum SubjectNameStrategy. Association-backed subjects work with Avro, generated Avro POCO, JSON Schema, Protobuf, schema references, and UseLatestVersion. Synchronous serialization fails fast until PrepareAsync or the relevant warmup API has completed. Producers perform this preparation during the first ProduceAsync for each topic and value shape; BuildAsync alone cannot resolve an association because it has neither.

Successful association and fallback resolutions are cached with a bounded cache. With FallbackStrategy = None, missing and failed lookups are not cached. A configured fallback remains cached even if an association is created later; call RefreshAsync to fetch and publish that association, Invalidate to evict one topic, or ClearCache after a broader governance change. Serializers and deserializers sharing that AssociatedNameStrategy invalidate their prepared subject entries automatically; an already-issued preparation admission remains bound to its original schema, while new messages require preparation against the refreshed subject.

Association endpoints require a Schema Registry deployment that supports the Confluent association API (Confluent Cloud or Confluent Platform 8.2+). Older or non-Confluent-compatible registries return an API error; configure a non-associated strategy or an explicit fallback for those environments.

Avro Serialization

With Generated Classes

using Dekaf.SchemaRegistry;
using Dekaf.SchemaRegistry.Avro;

var schemaRegistry = new CachedSchemaRegistryClient(
    new SchemaRegistryConfig { Url = "http://localhost:8081" }
);

var producer = await Kafka.CreateProducer<string, Order>()
    .WithBootstrapServers("localhost:9092")
    .WithValueSerializer(new AvroSerializer<Order>(schemaRegistry))
    .BuildAsync();

await producer.ProduceAsync("orders", order.Id, order);

With Generic Records

var serializer = new AvroSerializer<GenericRecord>(schemaRegistry);

var schema = (RecordSchema)Schema.Parse(@"{
    ""type"": ""record"",
    ""name"": ""Order"",
    ""fields"": [
        { ""name"": ""id"", ""type"": ""string"" },
        { ""name"": ""total"", ""type"": ""double"" }
    ]
}");

var record = new GenericRecord(schema);
record.Add("id", "order-123");
record.Add("total", 99.99);

await producer.ProduceAsync("orders", "order-123", record);

For zero-allocation GenericRecord serialization, Avro map values must use Dictionary<string, object>. Other IDictionary<string, object> implementations are rejected because their enumeration can allocate per message. Value-type arrays and lists are specialized for Avro primitives and built-in logical types; unsupported value-type element representations fail instead of silently boxing each element. Custom logical branches in unions must declare one sealed CLR type and exactly one effective value-dependent candidate for that type. Assignable or multi-candidate custom logical dispatch is rejected during writer construction because it would require a per-message candidate scan.

With source-generated POCOs

Dekaf.SchemaRegistry.Avro includes source-generated POCO support for plain CLR models that do not implement Apache Avro's ISpecificRecord. Opt in with [AvroRecord] on a top-level partial class, record, or struct. The package's bundled source generator emits the schema and strongly typed codec at build time; serialization uses constrained static dispatch with no reflection, boxing, runtime schema walk, or codec lookup.

using Dekaf.SchemaRegistry.Avro.Poco;

[AvroRecord(Name = "Order", Namespace = "example.orders")]
public sealed partial class Order
{
    [AvroField(Order = 0)]
    public required string Id { get; init; }

    [AvroField(Order = 1, Precision = 12, Scale = 2)]
    public decimal Total { get; init; }

    [AvroField(Order = 2, DefaultJson = "null")]
    public string? Note { get; init; }
}

using Dekaf;
using Dekaf.SchemaRegistry;
using Dekaf.SchemaRegistry.Avro.Poco;

using var registry = new SchemaRegistryClient(new SchemaRegistryConfig
{
    Url = "http://localhost:8081"
});

await using var producer = await Kafka.CreateProducer<string, Order>()
    .WithBootstrapServers("localhost:9092")
    .UseAvroPocoSchemaRegistry(registry)
    .BuildAsync();

Supported generated shapes are primitives, nullable members, enums, arrays, List<T>, Dictionary<string,T>, nested [AvroRecord] types, and explicit unions configured with UnionTypes. Only public fields and properties with public getters and setters are included; class inheritance is rejected so inherited state cannot be omitted silently. Explicit unions using an object or interface carrier accept reference-type branches only; value-type branches are rejected because assignment to the carrier would box on every read. Logical mappings are DateOnly → date, TimeOnly/TimeSpan → time-micros, DateTime/DateTimeOffset → timestamp-micros, Guid → uuid, and decimal → decimal. TimeSpan values must represent a time of day from zero through less than 24 hours. Generated DateTime fields must have Kind == DateTimeKind.Utc; local and unspecified values are rejected. Use DateTimeOffset when the source value carries an offset. Decimal members require Precision from 1 through 29 and Scale from 0 through the smaller of 28 and Precision.

Use Name, Aliases, and DefaultJson for schema evolution. Defaults must match the first Avro union branch; nullable fields therefore use DefaultJson = "null". Current generated defaults support null, primitive, string, bytes, and enum values. Invalid shapes, cycles, duplicate names/orders, ambiguous unions, and incompatible defaults fail compilation with DKAVRO diagnostics.

Call WarmupAsync before measuring or entering a latency-sensitive path. After warmup, serialization is 0 B per message for supported shapes. Deserialization allocates only the returned class/record and declared arrays, lists, dictionaries, strings, or byte arrays; cached schema-resolution plans add no per-message intermediate object graph. Rules remain opt-in and can allocate according to the configured rule executor. Generated and standard collection writers use one Avro block, allowing exact returned collection capacity. Valid external multi-block collections are also accepted and may grow their returned backing storage as later blocks arrive.

A generated POCO deserializer needs the writer schema ID from each record before it can prepare a reader plan. Call its WarmupAsync(schemaId) when schema IDs are known in advance. Otherwise, the first record for each unseen schema ID is prepared asynchronously by ConsumeAsync or ConsumeOneAsync; no consumer thread blocks on the Schema Registry request. Later records use the cached synchronous plan. Direct Deserialize calls and synchronous batch iteration cannot await a cold plan and fail fast, so call WarmupAsync(schemaId) before using ConsumeBatchAsync. The generated consumer convenience extension cannot pre-warm unknown writer schema IDs before the first record arrives.

Protobuf Serialization

using Dekaf.SchemaRegistry;
using Dekaf.SchemaRegistry.Protobuf;

var schemaRegistry = new CachedSchemaRegistryClient(
    new SchemaRegistryConfig { Url = "http://localhost:8081" }
);

var producer = await Kafka.CreateProducer<string, OrderProto>()
    .WithBootstrapServers("localhost:9092")
    .WithValueSerializer(new ProtobufSerializer<OrderProto>(schemaRegistry))
    .BuildAsync();

Protobuf imports use Schema Registry references by default. If you provide a custom ISchemaRegistryClient, implement LookupSchemaAsync; the serializer uses exact lookup to obtain the assigned version for every non-well-known imported .proto schema, including after automatic registration. The interface's default implementation throws NotSupportedException to identify custom clients that need updating. Set UseSchemaReferences = false only to retain the legacy registration behavior that omits references.

Known google/protobuf imports are skipped by default, matching Confluent's built-in dependency handling. This differs from earlier Dekaf releases. Set SkipKnownTypes = false if existing subjects must retain explicit references to those imports.

JSON Schema validation

JSON Schema Registry serialization does not validate payloads unless a JsonSchemaValidationOptions instance is supplied. Enable validation independently for writes, reads, or both:

using Dekaf.SchemaRegistry;
using Dekaf.SchemaRegistry.Json;

var registry = new SchemaRegistryClient(new SchemaRegistryConfig
{
    Url = "http://localhost:8081"
});

var validation = new JsonSchemaValidationOptions
{
    ValidatorFactory = new StreamingJsonSchemaValidatorFactory(registry),
    Mode = JsonSchemaValidationMode.Both
};

var producer = await Kafka.CreateProducer<string, Order>()
    .WithBootstrapServers("localhost:9092")
    .UseJsonSchemaRegistry(
        registry,
        orderSchema,
        jsonOptions: null,
        validationOptions: validation)
    .BuildAsync();

var consumer = await Kafka.CreateConsumer<string, Order>()
    .WithBootstrapServers("localhost:9092")
    .WithGroupId("orders")
    .UseJsonSchemaRegistry(
        registry,
        jsonOptions: null,
        validationOptions: validation)
    .BuildAsync();

JsonSchemaValidationMode.Serialize validates plaintext JSON after built-in domain rules and before encoding rules. Custom rule executors are validated before their transform because they do not expose phase boundaries. Deserialize validates after read rules and before JSON deserialization. This ordering lets domain rules establish the final logical shape and encoding rules transform the wire payload without validating ciphertext or a pre-migration shape.

Streaming validators are compiled once per exact registered Schema object and weakly cached. The serializer fetches the complete registered schema after registration or lookup, so write validation includes Schema Registry references. The deserializer compiles once when each schema ID is first encountered. References are resolved by subject and version with a configurable 30-second default timeout. Relative references resolve from the effective $id; internal JSON Pointer references are supported directly.

The allocation-free evaluator supports the common structural and scalar assertion subset of Draft 7, 2019-09, and 2020-12: types and nullability, object properties and required fields, additional properties, arrays and tuple items, size limits, and numeric limits. Unsupported assertion keywords fail during cold validator compilation instead of being silently ignored. Configure reference resolution and schema nesting limits when needed:

var factory = new StreamingJsonSchemaValidatorFactory(registry, new StreamingJsonSchemaValidatorOptions
{
    ReferenceResolutionTimeout = TimeSpan.FromSeconds(10),
    MaxSchemaDepth = 192
});

Invalid payloads throw JsonSchemaValidationException. SchemaId, Keyword, and JsonPath identify the failure. Exception messages never include payload contents. Validation has measurable CPU cost when enabled because each payload must be parsed and evaluated. Steady-state validation is zero-allocation; disabled serializers remain validation-neutral and do not load the optional JSON Schema package.

Migration rules

Set UseLatestVersion = true on a deserializer config to select the subject's latest registered schema as the reader schema. Dekaf resolves the writer's exact subject version, walks every adjacent version, and executes active migration rules before deserializing with the reader schema:

var rules = new SchemaRegistryRuleExecutor([migrationHandler]);

var config = new AvroDeserializerConfig
{
    UseLatestVersion = true,
    RuleExecutor = rules
};

var consumer = await Kafka.CreateConsumer<string, GenericRecord>()
    .WithBootstrapServers("localhost:9092")
    .WithGroupId("orders")
    .UseAvroSchemaRegistry(registry, config)
    .BuildAsync();

SchemaRegistryDeserializerConfig, AvroDeserializerConfig, and ProtobufDeserializerConfig all expose UseLatestVersion. Avro does not allow UseLatestVersion together with an explicit ReaderSchema.

Ordering matches Schema Registry behavior. Read encoding rules run against the writer schema first; upgrade or downgrade rules then run for each version edge; read domain rules run against the final reader schema last. The higher schema owns each edge's migration rules. Upgrade paths visit versions and rules in ascending/forward order. Downgrade paths visit versions and rules in descending/reverse order. UpDown rules participate in both directions, and paired success/failure actions select the first action for upgrade and the second for downgrade. Disabled rules are skipped.

Using the latest reader schema without active migration rules does not require a rule executor. If an active migration path exists, configure the built-in SchemaRegistryRuleExecutor; Dekaf fails closed instead of silently skipping the transform. Warm cached no-migration, disabled-migration, and active pass-through paths remain allocation-free, including interleaved writer schema IDs.

Migration plans follow SchemaRegistryConfig.LatestCacheTtlSecs. The Confluent-compatible default is -1, which disables time-based expiry. Set a non-negative TTL to periodically re-resolve latest schemas; 0 refreshes on every use. Historical version lookup includes deleted versions so migration paths remain complete. Custom ISchemaRegistryClient implementations must override the deleted-version overload; its default implementation fails closed.

JSONata rules

Install the optional Dekaf.SchemaRegistry.Jsonata package and register its handler when a data contract contains JSONATA rules:

using Dekaf.SchemaRegistry.Jsonata;

var rules = new SchemaRegistryRuleExecutor(
[
    new JsonataSchemaRegistryRuleHandler()
]);

The handler compiles and caches each rule expression, then evaluates JSONata against JSON codec payloads for write, read, and migration transforms. For example, $merge([$, {'fullName': first & ' ' & last}]) preserves the input object and adds fullName. JSONata dependencies remain isolated in the optional package; applications without the handler add no JSONata work or allocation.

Transform results may be any JSON value, including null, numbers, objects, and collections. JSONata's standard sequence semantics apply: a singleton sequence collapses to its value; append [] when the output must remain an array. An undefined result (for example, a missing-field query) fails explicitly rather than emitting invalid JSON. Condition rules must return true or false; false fails the rule. Invalid expressions, malformed JSON, and non-JSON payload formats fail with SchemaRegistryRuleException. Error messages identify the rule and engine error, but never include the payload.

Binary Avro and Protobuf codec payloads are not currently supported by the JSONata byte handler and are rejected explicitly. Their object-level transforms require codec-specific conversion before binary encoding.

Avro tagged-field encryption

Avro domain rules with type ENCRYPT transform only fields whose confluent:tags overlap the rule's tags. Tags may be declared directly on an Avro field or supplied through Schema Registry metadata using the field's fully qualified name. Tagged string fields store Base64 ciphertext; tagged bytes fields store raw ciphertext. Arrays, maps, nullable unions, nested records, and bytes-backed decimal logical types preserve their Avro shape while their tagged string or bytes values are transformed.

Fixed-width fields cannot hold variable-length ciphertext. Dekaf therefore rejects tagged Avro fixed fields, including fixed-backed decimal logical types, instead of rewriting the schema or silently encrypting the whole payload. Untagged fields remain byte-for-byte unchanged. After the schema and rule plan are cached, the field walker is allocation-free; the KMS provider still owns any algorithm- or key-management-specific costs.

For caller-owned mutable schemas, Avro metadata tag values must use FrozenSet<string> or IImmutableSet<string>. To update them, remove and re-add the containing metadata dictionary entry; Dekaf observes the dictionary's structural version and rebuilds the cached plan. Mutable HashSet<string> and SortedSet<string> metadata values are rejected because detecting their in-place changes would require an O(n) scan on every message. Rule tag sets may remain mutable; their version is checked once per transform.

Schema Registry Configuration

var config = new SchemaRegistryConfig
{
    Url = "http://localhost:8081",

    // Choose one authentication source:
    BasicAuthUserInfo = "username:password",
    BearerAuthToken = "eyJhbGciOi...",
    OAuthBearerConfig = oauthConfig,
    OAuthBearerTokenProvider = GetSchemaRegistryTokenAsync,

    // Optional mTLS client certificate
    ClientCertificate = certificate
};

var schemaRegistry = new CachedSchemaRegistryClient(config);

HTTP pipeline customization

SchemaRegistryClient accepts a caller-owned HttpMessageHandler, or a handler factory whose returned handler Dekaf owns. This supports custom tracing, retry, authentication, and policy handlers without adding a dependency on Microsoft.Extensions.Http:

var handler = new EnterprisePolicyHandler(new SocketsHttpHandler());
using var schemaRegistry = new SchemaRegistryClient(
    new SchemaRegistryConfig
    {
        Url = "https://schema-registry.example.com",
        UserAgent = "orders-service/2.1",
        DefaultHeaders = new Dictionary<string, string>
        {
            ["X-Tenant"] = "orders"
        }
    },
    handler);

Disposing SchemaRegistryClient never disposes a directly supplied HttpMessageHandler. The Func<HttpMessageHandler> overload transfers ownership of the returned handler to Dekaf. Authentication, timeout, failover, default headers, and the Schema Registry Accept header remain active around every custom pipeline. Content headers such as Content-Type cannot be configured as default request headers. Accept and User-Agent are also managed by Dekaf and cannot be supplied through DefaultHeaders. When UserAgent is not set, Dekaf sends a versioned Dekaf.SchemaRegistry/{version} value. RequestTimeoutMs must be positive, or -1 for an infinite timeout.

A custom handler owns proxy and TLS behavior completely. Therefore Tls, Proxy, UseProxy = false, and the legacy ClientCertificate property cannot be combined with a custom pipeline. The default pipeline supports the platform proxy or an explicit IWebProxy:

var config = new SchemaRegistryConfig
{
    Url = "https://schema-registry.example.com",
    Proxy = new WebProxy("http://proxy.example.com:8080")
};

Schema Registry TLS

The default pipeline supports caller-owned in-memory certificates and Dekaf-owned file or PEM material through SchemaRegistryTlsConfig:

var config = new SchemaRegistryConfig
{
    Url = "https://schema-registry.example.com",
    Tls = new SchemaRegistryTlsConfig
    {
        CaCertificatePath = "/run/secrets/schema-registry-ca", // PEM bundle or directory
        ClientCertificatePem = clientCertificatePem,
        ClientPrivateKeyPem = clientPrivateKeyPem,
        ClientCertificatePassword = privateKeyPassword,
        CheckCertificateRevocation = true
    }
};

Exactly one CA source and one client-certificate source may be configured. CA sources are a single certificate, a collection, a PEM string, or a file/directory path. Directories load only .pem, .crt, .cer, .pfx, and .p12 files, in ordinal path order; an empty directory or any malformed candidate fails client construction. PEM bundles load every certificate. Self-signed configured certificates and explicitly configured intermediate CAs are trust anchors. Server-provided intermediates are available to chain building without becoming trusted roots.

Client sources are an in-memory certificate with a private key, a PFX/P12 file, a PEM certificate plus separate key files, or PEM certificate/key strings. Encrypted PEM keys and password-protected PFX files use ClientCertificatePassword. Caller-provided certificate objects remain caller-owned; Dekaf disposes every certificate it loads. A client PFX/P12 may contain its intermediate chain; Dekaf retains and presents those intermediates with the leaf certificate. Passwords and private-key text are never included in validation exceptions.

TLS material is loaded once when the client is constructed and applies identically to every failover URL. To rotate file- or string-backed material, construct a new SchemaRegistryClient and dispose the old instance after in-flight operations complete. RemoteCertificateValidationCallback owns validation when set; otherwise ValidateServerCertificate, ValidateServerCertificateHostName, custom roots, revocation, and protocol settings are enforced by the default pipeline.

Google Cloud KMS

Install the opt-in Google Cloud provider when Schema Registry client-side field-level encryption (CSFLE) uses a Cloud KMS key:

dotnet add package Dekaf.SchemaRegistry.Kms.Gcp

using Dekaf.SchemaRegistry;
using Dekaf.SchemaRegistry.Kms.Gcp;

var gcpKms = new GcpKmsProvider();
var csfle = new SchemaRegistryCsfleRuleHandler(schemaRegistry, [gcpKms]);

The default constructor uses Google Application Default Credentials and the default Cloud KMS endpoint. Build and inject a client to select a regional endpoint, explicit credentials, emulator, or custom channel configuration:

using Google.Cloud.Kms.V1;

var kmsClient = new KeyManagementServiceClientBuilder
{
    Endpoint = "europe-west2-kms.googleapis.com"
}.Build();
var gcpKms = new GcpKmsProvider(kmsClient);

The supplied KeyManagementServiceClient is caller-owned and safe to share. Use the full CryptoKey resource name projects/<project>/locations/<location>/keyRings/<key-ring>/cryptoKeys/<key>. An optional gcp-kms:// prefix is accepted. Cloud KMS embeds the primary key version in its ciphertext, so the CryptoKey resource—not a CryptoKeyVersion—is used for both encryption and decryption.

Grant the runtime identity cloudkms.cryptoKeyVersions.useToEncrypt and cloudkms.cryptoKeyVersions.useToDecrypt, for example with the Cloud KMS CryptoKey Encrypter/ Decrypter role. Cancellation is forwarded to the gRPC call. Provider errors omit service response text and key material, and temporary SDK plaintext buffers are cleared after copy-out.

Alibaba Cloud KMS

Install the opt-in Alibaba Cloud provider only in applications whose Schema Registry client-side field-level encryption (CSFLE) keys are stored in Alibaba Cloud KMS:

dotnet add package Dekaf.SchemaRegistry.Kms.AliCloud

using Dekaf.SchemaRegistry;
using Dekaf.SchemaRegistry.Kms.AliCloud;

var aliCloudKms = new AliCloudKmsProvider();
var csfle = new SchemaRegistryCsfleRuleHandler(schemaRegistry, [aliCloudKms]);

Use KMS type alicloud-kms and a key URI in alicloud-kms://<region>/<key> format. Percent-encoded key paths are decoded for JVM compatibility; for example, alicloud-kms://cn-chengdu/alias%2Forders selects region cn-chengdu and key ID alias/orders. The ciphertext bytes use Confluent's format: the Alibaba Cloud CiphertextBlob is stored as UTF-8, while plaintext sent to and received from the KMS API is Base64.

The default constructor uses the Alibaba Cloud SDK credential chain. Prefer short-lived workload credentials. Explicit configuration can be supplied through AliCloudKmsProviderOptions, or per KEK through SchemaRegistryKmsKeyReference.KmsProps. Resolution order is KEK property, provider option, then environment variable. The provider accepts the current JVM property names below and the older Confluent .NET short names for compatibility.

Purpose	KEK property	Environment variable
KMS endpoint	alicloud.kms.endpoint	ALICLOUD_KMS_ENDPOINT
CA certificate file	alicloud.kms.caFile	ALICLOUD_KMS_CA_FILE
Credential type	alicloud.kms.credentialType	ALICLOUD_KMS_CREDENTIAL_TYPE
Access key ID	alicloud.kms.accessKeyId	ALIBABA_CLOUD_ACCESS_KEY_ID
Access key secret	alicloud.kms.accessKeySecret	ALIBABA_CLOUD_ACCESS_KEY_SECRET
STS security token	alicloud.kms.securityToken	ALIBABA_CLOUD_SECURITY_TOKEN
RAM role ARN	alicloud.kms.roleArn	ALICLOUD_KMS_ROLE_ARN, then ALIBABA_CLOUD_ROLE_ARN
Role session name	alicloud.kms.roleSessionName	ALICLOUD_KMS_ROLE_SESSION_NAME, then ALIBABA_CLOUD_ROLE_SESSION_NAME
Role session duration	alicloud.kms.roleSessionExpiration	ALICLOUD_KMS_ROLE_SESSION_EXPIRATION
Role policy	alicloud.kms.policy	ALICLOUD_KMS_ROLE_POLICY
STS endpoint	alicloud.kms.stsEndpoint	ALICLOUD_KMS_STS_ENDPOINT
External ID	alicloud.kms.externalId	ALICLOUD_KMS_EXTERNAL_ID

Credential type values are default, access_key, sts, and ram_role_arn; hyphens are also accepted. access_key requires an ID and secret. sts additionally requires a security token. ram_role_arn requires a role ARN and accepts either explicit source credentials or the default credential chain. Role sessions default to alicloud-kms-csfle; an explicit duration must be at least 900 seconds. Schema Registry default rule parameters using the rule.executors._default_.param. prefix are also recognized.

Endpoint and CA overrides affect every key using the matching resolved client configuration. The provider caches at most 64 thread-safe clients, so one instance can serve multiple regions without unbounded growth. IAliCloudKmsClientFactory supports application-controlled transports and credential brokers; clients returned by an injected factory remain caller-owned. Cancellation stops the caller's wait and is forwarded to injected clients; the upstream Alibaba Cloud SDK currently has no cancellation-token overload, so its underlying HTTP request may finish after cancellation.

Grant only the KMS permissions needed to encrypt and decrypt DEKs. Do not put credentials in key URIs, logs, or source control. Managed credential and service-response strings cannot be zeroed by .NET; keep them short-lived. Only trusted administrators should control KEK KMS properties: endpoint, CA, and credential overrides can redirect requests or disclose credentials. The package participates in trimming and NativeAOT analysis, and the injected-client path is AOT-safe. The default SDK path is also rooted by the NativeAOT smoke test; upstream SDK reflection behavior remains subject to the Alibaba Cloud SDK version in use.

HashiCorp Vault Transit KMS

Install the opt-in Vault provider when Schema Registry client-side field-level encryption (CSFLE) uses a Transit secrets-engine key:

dotnet add package Dekaf.SchemaRegistry.Kms.Vault

For token authentication, create one shared HttpClient and read the token from your secret-delivery mechanism, such as VAULT_TOKEN or a Vault Agent sink:

using Dekaf.SchemaRegistry;
using Dekaf.SchemaRegistry.Kms.Vault;

var httpClient = new HttpClient();
var tokenProvider = new VaultStaticTokenProvider(
    Environment.GetEnvironmentVariable("VAULT_TOKEN")!);
var transitClient = new VaultTransitHttpClient(httpClient, tokenProvider);
var vaultKms = new VaultKmsProvider(
    transitClient,
    vaultAddress: new Uri("https://vault.example:8200"),
    vaultNamespace: Environment.GetEnvironmentVariable("VAULT_NAMESPACE"));
var csfle = new SchemaRegistryCsfleRuleHandler(schemaRegistry, [vaultKms]);

For AppRole, replace the token provider. The provider logs in through the configured auth mount and caches each address/namespace token until shortly before its lease expires:

var tokenProvider = new VaultAppRoleTokenProvider(
    httpClient,
    roleId: Environment.GetEnvironmentVariable("VAULT_APPROLE_ROLE_ID")!,
    secretId: Environment.GetEnvironmentVariable("VAULT_APPROLE_SECRET_ID")!,
    authMountPoint: "approle");

Use KMS type hcvault and a Confluent-compatible key identifier in https://<vault-address>/<mount>/keys/<key-name> format, for example https://vault.example:8200/transit/keys/orders-kek. The provider extracts transit as the mount and orders-kek as the key name. For nested mounts, use a path such as https://vault.example:8200/team/transit/keys/orders-kek. The key identifier's scheme, host, and port must exactly match the locally configured vaultAddress; this prevents a Schema Registry KEK from redirecting Vault credentials to another server. Configure vaultAddress as a root authority without a path prefix. The namespace is sent as X-Vault-Namespace.

Grant update capability on <mount>/encrypt/<key> and <mount>/decrypt/<key>. The supplied HttpClient is caller-owned; VaultTransitHttpClient, VaultKmsProvider, and both token providers are safe for concurrent use. Cancellation reaches AppRole login and Transit HTTP requests. Errors omit Vault response bodies, key material, ciphertext, role IDs, secret IDs, and tokens. Serialized request/response buffers are zeroed before release. Credential and returned-token strings remain managed .NET strings and cannot be zeroed; source them from a secret-delivery mechanism and limit their lifetime accordingly.

AWS KMS for client-side field-level encryption

Install Dekaf.SchemaRegistry.Kms.Aws only in applications that use AWS KMS. The AWS SDK dependency stays out of Dekaf.SchemaRegistry and other serializer packages.

using Amazon;
using Dekaf.SchemaRegistry;
using Dekaf.SchemaRegistry.Kms.Aws;

using var awsKms = new AwsKmsProvider(RegionEndpoint.EUWest2);
var csfle = new SchemaRegistryCsfleRuleHandler(schemaRegistry, [awsKms]);
var rules = new SchemaRegistryRuleExecutor([csfle]);

AwsKmsProvider() uses the AWS SDK default credential and region provider chains. The region constructor fixes the KMS endpoint while retaining the default credential chain. For custom endpoints, retry settings, or other SDK options, pass an AmazonKeyManagementServiceConfig. For explicit credentials or application-managed client lifetimes, construct an AmazonKeyManagementServiceClient and pass it as IAmazonKeyManagementService; injected clients remain caller-owned unless ownsClient: true is specified.

The AWS SDK default credential chain checks explicitly configured client credentials first, then environment credentials, web-identity/container credentials, shared AWS profiles, and instance metadata as applicable to the host. Prefer short-lived workload credentials over long-lived access keys. The principal needs kms:Encrypt and kms:Decrypt for each configured key.

Schema Registry key references may contain a raw key ARN/alias or a Confluent-compatible aws-kms:// URI. Configure the provider's region or endpoint to match the key. The provider forwards cancellation to the AWS SDK, is safe for concurrent use, never logs key material or ciphertext, and clears temporary plaintext buffers where the runtime exposes them.

Each provider instance registers one KMS type. Applications using keys in multiple regions can give each regional provider a distinct type and use that type on the matching KEK:

using var euKms = new AwsKmsProvider(RegionEndpoint.EUWest2, type: "aws-kms-eu-west-2");
using var usKms = new AwsKmsProvider(RegionEndpoint.USEast1, type: "aws-kms-us-east-1");
var multiRegionCsfle = new SchemaRegistryCsfleRuleHandler(schemaRegistry, [euKms, usKms]);

Azure Key Vault KMS

Install the opt-in Azure provider when Schema Registry client-side field-level encryption (CSFLE) uses an Azure Key Vault key:

dotnet add package Dekaf.SchemaRegistry.Kms.Azure

using Azure.Identity;
using Dekaf.SchemaRegistry;
using Dekaf.SchemaRegistry.Kms.Azure;

var credential = new DefaultAzureCredential();
var azureKms = new AzureKeyVaultKmsProvider(credential);
var confluentAzureKms = new AzureKeyVaultKmsProvider(
    credential,
    type: AzureKeyVaultKmsProvider.ConfluentType);
var csfle = new SchemaRegistryCsfleRuleHandler(
    schemaRegistry,
    [azureKms, confluentAzureKms]);

DefaultAzureCredential uses the standard Azure credential chain. Production applications can instead pass a specific credential such as ManagedIdentityCredential or ClientSecretCredential. The credential and any supplied CryptographyClientOptions are caller-owned. For complete client-construction control, implement IAzureKeyVaultCryptographyClientFactory.

Use an absolute HTTPS key identifier with /keys/<name> or /keys/<name>/<version>, for example https://payments.vault.azure.net/keys/orders-kek. Azure public, US Government, and China Key Vault and Managed HSM DNS authorities are accepted; other authorities are rejected before credential use. Each provider instance registers one KMS type; register the default instance for azure-kv, the ConfluentType instance for Confluent-compatible azure-kms, or both as shown above. Matching azure-kv:// and azure-kms:// prefixes on the key identifier are optional. The provider uses RSA-OAEP-256. Prefer a versioned key identifier so existing data keeps decrypting after rotation. For a versionless key, set the KEK property encrypt.azure.key.version.save=true to embed the exact Azure key version in newly wrapped key material.

For RBAC-enabled vaults, grant the identity the Key Vault Crypto User role. For vaults using legacy access policies, grant the keys/wrapKey and keys/unwrapKey permissions. Managed HSM uses its own local RBAC system: grant the identity the Managed HSM Crypto User role at the /keys scope or the specific key's scope. One provider instance is safe for concurrent use. It bounds both its configured-key client cache and its ciphertext key-version client cache to 64 entries. Cancellation is forwarded to Azure; provider error messages do not include service response text or key material.

Consumer

using Dekaf;

var consumer = await Kafka.CreateConsumer<string, Order>()
    .WithBootstrapServers("localhost:9092")
    .WithGroupId("order-processors")
    .WithValueDeserializer(new AvroDeserializer<Order>(schemaRegistry))
    .SubscribeTo("orders")
    .BuildAsync();

await foreach (var msg in consumer.ConsumeAsync(ct))
{
    Order order = msg.Value;
    // Process order
}

Schema Evolution

Schema Registry handles schema evolution:

// V1: Original schema
public class OrderV1
{
    public string Id { get; set; }
    public decimal Total { get; set; }
}

// V2: Added field with default (backward compatible)
public class OrderV2
{
    public string Id { get; set; }
    public decimal Total { get; set; }
    public string Status { get; set; } = "pending";  // New field with default
}

The serializer automatically registers new schema versions and handles compatibility.

For GenericRecord, the serializer keys writers, subjects, and schema IDs by the runtime Avro schema's logical identity. Equivalent schema instances reuse one cache entry, while different versions on the same TopicName subject retain their own IDs. Runtime-schema caches are bounded; configure the positive limit when applications intentionally produce many distinct schemas:

var serializer = new AvroSchemaRegistrySerializer<GenericRecord>(schemaRegistry, new AvroSerializerConfig
{
    MaxCachedSchemas = 500 // Default: 1000
});

After the primary strong-cache limit is reached, exact schema objects use weak-key entries and a second FIFO logical cache retains a bounded overflow working set. The overflow cache uses the same configured limit, with a minimum of three entries for short schema rotations. A two-entry hot set still gives repeated and alternating overflow schemas a lock-free reference fast path. Specific records share one stateless writer; generic records retain a writer per logical schema. Cache entries never retain individual GenericRecord values.

Reuse parsed Avro Schema objects on the per-message path. A previously observed schema object uses the O(1) reference lookup. A newly parsed object is a first-seen identity even when it is logically equivalent to an existing schema; safely proving that equivalence requires a structural fingerprint or comparison. Parse and cache schemas during startup or producer setup instead of constructing a new runtime schema for every message.

Each first-seen overflow schema identity gets a weak association with its logical cache entry. This adds metadata only on the cold first-seen path, does not retain the schema object, and lets any live equivalent instance reuse its writer after logical-cache and hot-set eviction. Repeated use of an observed schema identity remains allocation-free.

Subject Naming Strategies

var serializer = new AvroSchemaRegistrySerializer<Order>(schemaRegistry, new AvroSerializerConfig
{
    SubjectNameStrategy = SubjectNameStrategy.TopicRecordName
});

Strategy	Value or key subject
TopicName	{topic}-value or {topic}-key
RecordName	{fully-qualified-record-name}
TopicRecordName	{topic}-{fully-qualified-record-name}

These formats match Confluent serializers. Avro GenericRecord subjects use the fullname from the record's runtime schema, JSON Schema subjects use the schema title when present, and Protobuf subjects use the message descriptor's full name.

Avro-generated ISpecificRecord types serialize without per-message allocations when their record fields are scalar null, boolean, int, long, float, double, string, or bytes fields exposed by matching public properties. Unsupported SpecificRecord shapes fail when the serializer is created instead of silently falling back to Apache Avro's allocating Get(int): object path. Use GenericRecord for collection, union, enum, fixed, logical, or nested record fields.

When using the generic SchemaRegistrySerializer with a record-based strategy, prefer its subject-independent Func<Schema> schema factory overload. The subject-aware Func<string, Schema> overload may be called again with the schema-derived subject until the callback and schema name agree.

Migrating subjects created before this fix

Older Dekaf releases appended -key or -value to RecordName and TopicRecordName subjects. Before upgrading producers, register or copy each schema version from the old suffixed subject to the standard subject. Keep the same compatibility mode and version order. For example:

• com.example.Order-value becomes com.example.Order.
• orders-com.example.Order-key becomes orders-com.example.Order.

If consumers or deployment sequencing require a gradual migration, keep the old names temporarily:

var config = new AvroSerializerConfig
{
    SubjectNameStrategy = SubjectNameStrategy.RecordName,
    UseLegacySubjectNames = true
};

UseLegacySubjectNames is also available on ProtobufSerializerConfig and as an optional constructor/builder-extension argument for JSON Schema and generic Schema Registry serializers. It affects only the enum-based RecordName and TopicRecordName strategies; TopicName and custom ISubjectNameStrategy implementations are unchanged. Disable the option after every producer and schema has moved to the standard subject.

Schema identity framing and Confluent interoperability

Dekaf reads and writes the Confluent Schema Registry identity framing used by Avro, JSON Schema, and Protobuf serializers. Prefix framing places magic byte 0 and the big-endian global schema ID before the payload. Header framing leaves the payload unprefixed and writes magic byte 1 plus the network-order schema GUID to __key_schema_id or __value_schema_id. Protobuf appends its message index data to that identity header.

The serializer and deserializer strategies are intentionally separate:

Writer / reader strategy	Prefix record	Header record	Behavior
SchemaIdSerializerStrategy.Prefix	Writes	—	Confluent-compatible default; payload carries the global integer ID.
SchemaIdSerializerStrategy.Header	—	Writes	Payload is unprefixed; a Headers collection is required.
SchemaIdDeserializerStrategy.Prefix	Reads	Header ignored	Parses the raw payload as a prefix without inspecting identity headers.
SchemaIdDeserializerStrategy.Header	Rejects	Reads	Requires the reserved key/value identity header.
SchemaIdDeserializerStrategy.Dual	Reads	Reads	Header wins when present; otherwise falls back to the prefix. This is the default reader strategy.

Dual is a reader migration mode, not a serializer mode. Do not combine a prefixed payload and a GUID header into one record: when the header is present, Dual treats byte zero as application payload. A malformed reserved header never falls back to a valid prefix. Duplicate reserved headers follow Kafka header conventions: the last matching key/value identity is authoritative.

Prefix is not a strict guard against header-framed records. It ignores identity headers and parses the first five payload bytes as a prefix. If those bytes happen to contain magic byte 0 and a registered nonnegative schema ID, the reader can decode the remaining bytes under an unintended schema instead of rejecting the record.

For a rolling migration, deploy Dual readers first, switch writers from Prefix to Header, then optionally tighten readers to Header after all prefixed records have aged out. Keep Dual when a topic deliberately accepts both writer generations.

using Dekaf.SchemaRegistry;
using Dekaf.SchemaRegistry.Avro;

var writer = new AvroSchemaRegistrySerializer<Order>(
    registry,
    new AvroSerializerConfig
    {
        // Use an already registered schema and emit its GUID as a Kafka header.
        UseSchemaId = 123,
        AutoRegisterSchemas = false,
        SchemaIdStrategy = SchemaIdSerializerStrategy.Header
    });

var reader = new AvroSchemaRegistryDeserializer<Order>(
    registry,
    new AvroDeserializerConfig
    {
        // Accept existing prefix records and newly written header records.
        SchemaIdStrategy = SchemaIdDeserializerStrategy.Dual
    });

UseSchemaId has precedence over UseLatestVersion and AutoRegisterSchemas. Dekaf fetches the selected schema and validates that its format, root type, and Protobuf reference graph match the serializer type before emitting any bytes. UseLatestVersion takes precedence over auto-register or lookup when no explicit ID is set. Otherwise AutoRegisterSchemas = true registers or reuses the schema; false performs exact lookup.

Header serialization resolves the registered schema GUID during preparation and caches the encoded frame. Header deserialization resolves GUIDs through Schema Registry; warm serializers and deserializers reuse bounded caches. The registry must return a non-empty GUID and support GUID lookup. The checked-in interoperability vectors are generated with Confluent Schema Registry SerDes 2.15.0 and confluentinc/cp-schema-registry:8.2.0 for all three schema formats.

Within the framing selected by the reader strategy, identity failures are deterministic:

• Missing, null, empty, truncated, or unknown-magic headers fail instead of falling back.
• Negative, truncated, or unknown-magic prefix IDs fail before payload decoding.
• A GUID resolving to the wrong schema format or an explicit ID resolving to the wrong record type fails before serialization/deserialization completes.
• If subject-scoped lookup returns a different GUID or ID than the identity being processed, Dekaf reports the conflict; it does not silently select either schema.
• Protobuf rejects malformed or trailing message-index data after the GUID frame.

Cross-client rollout tests should cover both key and value header names, every schema format, old prefix records, new header records, and malformed reserved headers. Do not strip the reserved identity header in middleware or copy it between key and value components.