Thesis Schema Registry

A PHP client for Confluent Schema Registry with full coverage of the REST API. Format-agnostic by design: works with Avro, JSON Schema, and Protobuf schemas alike.

Why

Schema Registry stores versioned schemas and enforces compatibility rules between them. It's the central piece behind Kafka pipelines that need to evolve message formats without breaking producers or consumers. This library is the HTTP client for that registry — it does not parse Avro or any other format itself, it just speaks the REST protocol.

Installation

composer require thesis/schema-registry

HTTP transport

The client is built on PSR-18, so any compliant HTTP client works. For non-blocking I/O — especially when running alongside thesis/kafka — the recommended driver is amphp/http-client-psr7, which adapts the Fiber-based amphp HTTP client to PSR-18.

composer require amphp/http-client-psr7 guzzlehttp/psr7

guzzlehttp/psr7 is used here only as a PSR-7 message factory; any PSR-7 implementation will do.

Creating a client

use Amp\Http\Client\HttpClientBuilder;
use Amp\Http\Client\Psr7\PsrAdapter;
use Amp\Http\Client\Psr7\PsrHttpClient;
use GuzzleHttp\Psr7\HttpFactory;
use Thesis\SchemaRegistry;

$psrHttpFactory = new HttpFactory();

$client = new SchemaRegistry\Client(
    http: new PsrHttpClient(
        HttpClientBuilder::buildDefault(),
        new PsrAdapter($psrHttpFactory, $psrHttpFactory),
    ),
    requestFactory: $psrHttpFactory,
    streamFactory: $psrHttpFactory,
    config: new SchemaRegistry\Config(
        uri: new SchemaRegistry\URI('http://schema-registry:8081'),
    ),
);

Configuration

The Config object accepts the registry URI plus optional user agent, authentication, and a default set of Parameters applied to every request:

use Thesis\SchemaRegistry;
use Thesis\SchemaRegistry\Authentication;
use Thesis\SchemaRegistry\Parameter;

$config = new SchemaRegistry\Config(
    uri: new SchemaRegistry\URI('http://schema-registry:8081'),
    userAgent: 'my-service/1.0',
    authentication: new Authentication\Basic('key', 'secret'),
    parameters: SchemaRegistry\Parameters::of(
        new Parameter\InContext('production'),
    ),
);

Using the client

All methods on Client are typed wrappers over individual REST endpoints. Most calls accept an optional variadic Parameter ...$parameters to tweak per-call behaviour (filters, normalization, context overrides, etc.) — see PHPDoc on each method for the list of applicable parameters.

Registering and fetching schemas

use Thesis\SchemaRegistry;

$id = $client->registerSchema(
    subject: 'orders-value',
    schema: new SchemaRegistry\Schema('{"type":"record",...}', SchemaRegistry\SchemaType::Avro),
);

$schema = $client->schemaByVersion('orders-value', SchemaRegistry\Version::latest());
$text = $client->schemaTextByVersion('orders-value', SchemaRegistry\Version::latest());
$all = $client->schemas('orders-value'); // every version, ordered

Registering the same schema string under the same subject returns the existing id — there's no need to deduplicate on the caller side.

Listing

$subjects = $client->subjects();
$versions = $client->subjectVersions('orders-value');
$allSchemas = $client->allSchemas();
$supported = $client->supportedTypes(); // [Avro, Protobuf, JSON]

Lookup by id

A schema id is global across the registry. To find where it's used:

$subjects = $client->subjectsById($id)
$versions = $client->schemaVersionsById($id);
$schemas = $client->schemaUsages($id);

Lookup by schema content

lookupSchema checks whether a schema is already registered under a subject without creating a new version:

$existing = $client->lookupSchema('orders-value', $schema);

Compatibility

use Thesis\SchemaRegistry;

$result = $client->checkCompatibility('orders-value', $newSchema);
if (!$result->isCompatible) {
    // ...
}

$global = $client->compatibility();
$subjectLevel = $client->compatibility('orders-value');

$client->setCompatibility(new SchemaRegistry\Compatibility(new SchemaRegistry\CompatibilityLevel::Backward), 'orders-value');
$client->resetCompatibility('orders-value');

Deletion

By default, deletes are soft — the schema disappears from regular listings but remains in storage. Pass Parameter\Permanent for a hard delete (a version must be soft-deleted first):

use Thesis\SchemaRegistry\Parameter;

$client->deleteSubject('orders-value');
$client->deleteSubject('orders-value', new Parameter\Permanent(true));

$client->deleteSchema('orders-value', 1);

References

Schemas can reference each other. To discover what depends on a given version:

use Thesis\SchemaRegistry;

$dependents = $client->schemaReferences('common-types', SchemaRegistry\Version::latest());

Contexts

Schema Registry supports multiple isolated namespaces called contexts. Set one as the default on the config:

use Thesis\SchemaRegistry;
use Thesis\SchemaRegistry\Parameter;

$config = new SchemaRegistry\Config(
    uri: new SchemaRegistry\URI('http://schema-registry:8081'),
    parameters: SchemaRegistry\Parameters::of(
        new Parameter\InContext('production'),
    ),
);

Or override per call:

$client->registerSchema(
    'orders-value',
    $schema,
    parameters: [new Parameter\InContext('staging')],
);

The per-call value replaces the default for that request; without an override, requests go to the configured context (or the global namespace if none is set).

Authentication

Several authentication schemes are supported. Construct the appropriate Authentication and pass it to Config:

use Thesis\SchemaRegistry\Authentication;
// Basic — also used for Confluent Cloud (API key as username, secret as password)
new Authentication\Basic('key', 'secret');

// Bearer token (OAuth 2.0)
new Authentication\Bearer($token);

// mTLS is configured on the HTTP client itself, not on the registry client

Low-level access

If an endpoint isn't covered by a typed method, drop down to Client::call():

$result = $client->call(new MyCustomRequest(), new Parameter\Normalize());

This is the same primitive every high-level method is built on.