feat(schema-registry): manage global and subject compatibility levels

[thomhurst]

Problem

ISchemaRegistryClient can test whether a candidate schema is compatible, but cannot read or update Schema Registry compatibility configuration. Applications, deployment tools, and tests therefore need another HTTP client to manage global or subject-specific compatibility policy.

Missing operations:

• Read global compatibility level (GET /config).
• Update global compatibility level (PUT /config).
• Read subject compatibility level (GET /config/{subject}).
• Update subject compatibility level (PUT /config/{subject}).

Related Confluent proposal: confluentinc/confluent-kafka-dotnet#2111

Relevant Dekaf code:

• src/Dekaf.SchemaRegistry/ISchemaRegistryClient.cs — currently exposes IsCompatibleAsync only.
• src/Dekaf.SchemaRegistry/SchemaRegistryClient.cs — existing failover, authentication, cancellation, source-generated JSON, and error handling patterns.
• src/Dekaf.SchemaRegistry/SchemaRegistryJsonContext.cs — AOT-safe request/response serialization registrations.

Proposed API

public enum SchemaCompatibilityLevel
{
    None,
    Backward,
    BackwardTransitive,
    Forward,
    ForwardTransitive,
    Full,
    FullTransitive
}

Task<SchemaCompatibilityLevel> GetCompatibilityAsync(
    string? subject = null,
    CancellationToken cancellationToken = default);

Task<SchemaCompatibilityLevel> UpdateCompatibilityAsync(
    SchemaCompatibilityLevel compatibility,
    string? subject = null,
    CancellationToken cancellationToken = default);

subject: null means global configuration. Empty or whitespace subject should fail fast rather than ambiguously targeting global configuration; only null selects global scope.

If supported target Schema Registry versions expose additional values, model them explicitly after verifying wire names and server behavior. Unknown future values should produce a diagnostic SchemaRegistryException, not silently map to a different policy.

Wire mapping

Use exact Schema Registry strings:

• NONE
• BACKWARD
• BACKWARD_TRANSITIVE
• FORWARD
• FORWARD_TRANSITIVE
• FULL
• FULL_TRANSITIVE

Expected endpoints:

• Global: config
• Subject: config/{Uri.EscapeDataString(subject)}

Request and response DTOs must match Schema Registry's compatibility / compatibilityLevel JSON contract as applicable. Register DTOs in SchemaRegistryJsonContext; no reflection-based serialization.

Implementation approach

1. Add public enum and interface methods with XML documentation describing global/subject semantics.
2. Add internal source-generated request/response DTOs. Centralize enum-to-wire conversion and strict wire-to-enum parsing.
3. Reuse existing base-URL failover, authentication, timeout, cancellation, and EnsureSuccessAsync behavior.
4. Escape subject as one URI path segment using the same conventions as other subject APIs.
5. Validate arguments before network I/O.
6. Preserve server error code/message in SchemaRegistryException, including 404 for missing subject config and 422 for invalid configuration.
7. Do not cache compatibility configuration initially. It is mutable administrative state; hidden caching would create stale reads. Add caching only later with explicit consistency semantics.
8. Return the compatibility level acknowledged by the server, not merely the requested value.
9. Ensure every await in library code uses ConfigureAwait(false).

Tests

Unit HTTP tests

• Global GET uses /config and parses every supported value.
• Subject GET escapes special characters correctly.
• Global and subject PUT serialize exact wire values.
• PUT returns server-acknowledged value.
• Null subject selects global; empty/whitespace subject throws.
• Unknown response value fails clearly.
• Authentication headers, base URL failover, cancellation, and timeout follow existing client behavior.
• Structured Schema Registry errors retain status/error code/message.
• Source-generated JSON works under trimming/NativeAOT.

Integration

Using Schema Registry Testcontainers:

1. Save original global compatibility level.
2. Update global level and verify readback.
3. Register a unique subject.
4. Update subject level and verify it differs from global without changing global.
5. Exercise at least one transitive mode.
6. Restore global state in finally so parallel/failed tests do not contaminate later tests.

Mark configuration-mutating tests non-parallel or isolate Schema Registry instances.

NativeAOT

Extend Schema Registry AOT smoke coverage to call GET and PUT compatibility operations, proving DTO metadata is rooted.

Acceptance criteria

• Public API reads and updates global and subject compatibility configuration.
• All standard compatibility modes round-trip exactly.
• Subject paths are safely escaped and argument semantics documented.
• Existing failover/authentication/cancellation/error behavior is reused.
• No stale implicit cache is introduced.
• Unit, integration, and NativeAOT tests pass.
• No producer/consumer hot-path changes or performance regression.

Non-goals

• Do not conflate configuration management with IsCompatibleAsync schema testing.
• Do not add per-message compatibility checks.
• Do not introduce a general-purpose raw REST escape hatch.