Package com.aim2be.platform.schema

Interface SchemaRegistryClient

All Known Implementing Classes:
ApicurioSchemaRegistryClient, ConfluentWireCompatSchemaRegistryClient, DualSchemaRegistryClient
public interface SchemaRegistryClient
Schema-registry abstraction (ADR-0011 §B).

The aim2be Kafka substrate runs Apicurio Registry as the primary authority for Avro / Protobuf / JSON Schema definitions. A Confluent wire-format compatibility path is provided so consumers that resolved against the Confluent registry before the Apicurio migration keep working unchanged. Both paths share this interface — both register-by-subject and fetch-by-globalId calls behave identically across implementations; the divergence is in the on-wire byte prefix selected for each emitted record (see WireFormat).

Implementations:

  • ApicurioSchemaRegistryClient — native path, emits the Apicurio 8-byte big-endian globalId prefix. Activated when im2be.schema-registry.mode=apicurio (default).
  • ConfluentWireCompatSchemaRegistryClient — wire-format alternate; talks to the SAME Apicurio registry but encodes the 0x00 magic-byte + 4-byte big-endian schemaId Confluent prefix. Activated when im2be.schema-registry.mode=confluent.
  • DualSchemaRegistryClient — wire-format coexistence coordinator. NOT an HA hedge: both delegates wrap the same underlying registry, so any outage propagates from the primary to the caller (a retry through the alternate-prefix delegate cannot recover from a real outage). The dual coordinator exists so a deployment can simultaneously offer both 8-byte (Apicurio) and 4-byte (Confluent) wire formats against one Apicurio backend — useful during phased migrations. Activated when im2be.schema-registry.mode=dual.

The mode is selected at autoconfig time; exactly ONE bean implementing this interface is active in any given application context (see SchemaRegistryAutoConfiguration for the conditional wiring).

  • Method Details

    • register

      long register(String subject, byte[] schemaBytes, SchemaRegistryClient.SchemaType type)
      Registers schemaBytes under subject, returning the registry-assigned global ID.

      Idempotent: a re-registration of identical bytes returns the same global ID (Apicurio's IfExists=RETURN_OR_UPDATE semantics).

      Parameters:
      subject - the subject name (typically <topic>-value or <topic>-key per ADR-0011 §B)
      schemaBytes - the schema body bytes; UTF-8 JSON for Avro/JSON, binary for Protobuf; must be non-null
      type - the schema type (AVRO / PROTOBUF / JSON)
      Returns:
      the registry-assigned global ID

    • fetchByGlobalId

      SchemaRegistryClient.Schema fetchByGlobalId(long globalId)
      Fetches a schema by registry global ID.
      Parameters:
      globalId - the global ID returned by a prior register(java.lang.String, byte[], com.aim2be.platform.schema.SchemaRegistryClient.SchemaType) call
      Returns:
      the schema record, or null if no such globalId. Sentinel-field contract: the Apicurio v2 REST API has no direct globalId → (subject, version) lookup, so impls that hydrate via the content-by-globalId endpoint return a SchemaRegistryClient.Schema with subject="" and version=0 as sentinels. Callers that need subject/version coordinates must use fetchLatestBySubject(String) instead.

    • fetchLatestBySubject

      SchemaRegistryClient.Schema fetchLatestBySubject(String subject)
      Fetches the latest schema body registered under subject.
      Parameters:
      subject - the subject name
      Returns:
      the schema record, or null if no such subject

    • kind

      SchemaRegistryClient.Kind kind()
      Returns the implementation kind. Used for observability tagging (the kind shows up as schema.registry on every OTel span this client emits) and for callers that need to choose between Apicurio globalId semantics and Confluent schemaId semantics in their producer code path.
      Returns:
      the implementation kind; never null

    • wireFormat

      default WireFormat wireFormat()
      Returns the wire format this client emits / parses. Defaults to WireFormat.APICURIO_8_BYTE — the Apicurio native prefix. ConfluentWireCompatSchemaRegistryClient overrides this to WireFormat.CONFLUENT_4_BYTE.
      Returns:
      the wire format; never null