Package com.aim2be.platform.schema

Class ConfluentWireCompatSchemaRegistryClient

java.lang.Object

com.aim2be.platform.schema.ConfluentWireCompatSchemaRegistryClient

All Implemented Interfaces:

SchemaRegistryClient

public class ConfluentWireCompatSchemaRegistryClient
extends Object
implements SchemaRegistryClient

Confluent wire-format compatibility client (ADR-0011 §B fallback hedge).

Talks to an Apicurio registry under the hood — same v2 REST API, same artifact storage — but advertises WireFormat.CONFLUENT_4_BYTE as its on-wire prefix. Producers built on top of this client emit records using Confluent's 0x00 magic + 4-byte schemaId layout instead of Apicurio's 8-byte globalId prefix.

This is the "Confluent if missed" hedge from .planning/26-stage-b-outbox-parity.md §0 row L-19: if ADR-0011 §B is overridden later and aim2be needs to stay wire-compatible with existing Confluent consumers, switching every producer is a 1-bean config change (im2be.schema-registry.mode=confluent) rather than a code rewrite.

Behaviour delta vs ApicurioSchemaRegistryClient:

• kind() returns SchemaRegistryClient.Kind.CONFLUENT_WIRE_COMPAT
• wireFormat() returns WireFormat.CONFLUENT_4_BYTE
• register / fetch operations delegate verbatim to the Apicurio impl — the storage backend is the same, only the producer-side prefix chosen at emit time differs

Note: Apicurio's globalId is a long and Confluent's schemaId is an int. The platform's choice (per ADR-0011 §B) is to widen the Confluent id to long at the SchemaRegistryClient boundary; downstream producer encoders narrow back to int via WireFormatCodec.prependConfluentPrefix(int, byte[]) after a range check. Operators must keep the registry's id allocator within the 31-bit positive int range during the period this mode is active.

Nested Class Summary

Nested classes/interfaces inherited from interface com.aim2be.platform.schema.SchemaRegistryClient

SchemaRegistryClient.Kind, SchemaRegistryClient.Schema, SchemaRegistryClient.SchemaType

Constructor Summary

Constructors

Constructor	Description
ConfluentWireCompatSchemaRegistryClient(io.apicurio.registry.rest.client.RegistryClient delegate, String groupId, String compatibility, io.opentelemetry.api.OpenTelemetry openTelemetry)

Method Summary

Modifier and Type	Method	Description
SchemaRegistryClient.Schema	fetchByGlobalId(long globalId)	Fetches a schema by registry global ID.
SchemaRegistryClient.Schema	fetchLatestBySubject(String subject)	Fetches the latest schema body registered under subject.
SchemaRegistryClient.Kind	kind()	Returns the implementation kind.
long	register(String subject, byte[] schemaBytes, SchemaRegistryClient.SchemaType type)	Registers schemaBytes under subject, returning the registry-assigned global ID.
WireFormat	wireFormat()	Returns the wire format this client emits / parses.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Details

ConfluentWireCompatSchemaRegistryClient

public ConfluentWireCompatSchemaRegistryClient(io.apicurio.registry.rest.client.RegistryClient delegate,
 String groupId,
 String compatibility,
 io.opentelemetry.api.OpenTelemetry openTelemetry)

Parameters:

delegate - the Apicurio v2 REST client; not null

groupId - the artifact group ID; not null

compatibility - compatibility rule (BACKWARD/FORWARD/FULL/NONE); forwarded to the inner Apicurio delegate. May be null/blank for no enforcement.

openTelemetry - the OpenTelemetry instance; not null

Method Details

register

public long register(String subject,
 byte[] schemaBytes,
 SchemaRegistryClient.SchemaType type)

Description copied from interface: SchemaRegistryClient

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

Specified by:

register in interface SchemaRegistryClient

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

public SchemaRegistryClient.Schema fetchByGlobalId(long globalId)

Description copied from interface: SchemaRegistryClient

Fetches a schema by registry global ID.

Specified by:

fetchByGlobalId in interface SchemaRegistryClient

Parameters:

globalId - the global ID returned by a prior SchemaRegistryClient.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 SchemaRegistryClient.fetchLatestBySubject(String) instead.

fetchLatestBySubject

public SchemaRegistryClient.Schema fetchLatestBySubject(String subject)

Description copied from interface: SchemaRegistryClient

Fetches the latest schema body registered under subject.

Specified by:

fetchLatestBySubject in interface SchemaRegistryClient

Parameters:

subject - the subject name

Returns:

the schema record, or null if no such subject

kind

public SchemaRegistryClient.Kind kind()

Description copied from interface: SchemaRegistryClient

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.

Specified by:

kind in interface SchemaRegistryClient

Returns:

the implementation kind; never null

wireFormat

public WireFormat wireFormat()

Description copied from interface: SchemaRegistryClient

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.

Specified by:

wireFormat in interface SchemaRegistryClient

Returns:

the wire format; never null