> ## Documentation Index
> Fetch the complete documentation index at: https://docs.streamnative.io/llms.txt
> Use this file to discover all available pages before exploring further.

# Connect to the Kafka Schema Registry

> Find your Schema Registry endpoint, authenticate, and configure Kafka clients to register and resolve schemas.

To use the Kafka Schema Registry you need three things: its URL, credentials, and permission on the
subjects your application touches.

## Prerequisites

* A Kafka cluster, or a Pulsar cluster with the Kafka protocol enabled, on StreamNative Cloud.
* A service account with an API key. See
  [Create an API key](/cloud/security/authentication/service-accounts/use-api-keys/api-keys-overview#create-an-api-key).
* Schema Registry permissions for that service account, granted through role-based access
  control (RBAC). See [Grant access](#grant-access).

## Get the Schema Registry URL

The URL depends on the cluster type.

<Tabs>
  <Tab title="Kafka cluster">
    The Schema Registry is served at the cluster's HTTP service URL, with no sub-path:

    ```
    https://<kafka-cluster-dns-name>
    ```

    ```shell theme={null}
    snctl get schemaregistry <cluster-name> \
      -o jsonpath='https://{.status.serviceEndpoints[?(@.type=="external")].dnsName}{"\n"}'
    ```
  </Tab>

  <Tab title="Pulsar cluster">
    On a Pulsar cluster with the Kafka protocol enabled, the URL is the cluster's HTTP service URL with
    **`/kafka` appended**:

    ```
    https://<pulsar-cluster-dns-name>/kafka
    ```

    ```shell theme={null}
    snctl get pulsarcluster <cluster-name> \
      -o jsonpath='https://{.spec.serviceEndpoints[?(@.type=="service")].dnsName}/kafka{"\n"}'
    ```
  </Tab>

  <Tab title="Cloud Console">
    1. Go to the **Cluster Details** page.
    2. Find the **HTTP Service URL** in the cluster dashboard.
    3. On a Pulsar cluster, append `/kafka`.
  </Tab>
</Tabs>

<Warning>
  The `/kafka` sub-path applies only to Pulsar clusters. A URL copied from one cluster type won't work
  on the other.
</Warning>

The registry listens on port 443 over HTTPS. See [Networking](/cloud/networking/networking) for all
cluster endpoints and ports.

## Authenticate

Three mechanisms are available.

### Basic authentication

Works with every Kafka client. Supply your API key as the password; the username can be any non-empty
string.

```java theme={null}
props.put(KafkaAvroSerializerConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl);
props.put(KafkaAvroSerializerConfig.BASIC_AUTH_CREDENTIALS_SOURCE, "USER_INFO");
props.put(KafkaAvroSerializerConfig.USER_INFO_CONFIG, String.format("%s:%s", "any-user", apiKey));
```

<Note title="Note">
  Only the password is used as the credential. The username is ignored, so pass any non-empty
  placeholder.
</Note>

### OAuth2

Available for the Kafka Java client. Add the `oauth-client` dependency alongside your Kafka client
and serializer:

```xml theme={null}
<dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka-clients</artifactId>
  <version>3.6.1</version>
</dependency>
<dependency>
  <groupId>io.streamnative.pulsar.handlers</groupId>
  <artifactId>oauth-client</artifactId>
  <version>3.2.2.6</version>
</dependency>
<dependency>
  <groupId>io.confluent</groupId>
  <artifactId>kafka-avro-serializer</artifactId>
  <version>7.5.0</version>
</dependency>
```

Minimum versions: `kafka-clients` 3.4.0, `oauth-client` 3.1.0.4, `kafka-avro-serializer` 7.5.0.

<Note title="Note">
  Before version 3.2.2.6, `oauth-client` requires Java 17 or later.
</Note>

On top of the properties that already configure OAuth2 for the Kafka connection, add:

```java theme={null}
props.put(KafkaAvroSerializerConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl);
props.put(KafkaAvroSerializerConfig.BEARER_AUTH_CUSTOM_PROVIDER_CLASS,
    "io.streamnative.pulsar.handlers.kop.security.oauth.schema.OauthCredentialProvider");
props.put(KafkaAvroSerializerConfig.BEARER_AUTH_CREDENTIALS_SOURCE, "CUSTOM");
```

See [Configure Kafka clients with OAuth 2.0](/cloud/security/authentication/service-accounts/use-oauth/configure-kafka-clients-with-oauth-20)
for the Kafka-connection half of the configuration.

### mTLS

When the cluster presents a TLS listener and mutual TLS is configured, the client certificate
identifies the caller and the `Authorization` header is ignored.

## Grant access

Assign one of the Schema Registry roles, scoped to the subjects the client uses:

| Role                                                                             | Grants                                                                 |
| -------------------------------------------------------------------------------- | ---------------------------------------------------------------------- |
| [`schema-reader`](/cloud/security/access/rbac/manage-rbac-roles#schema-reader)   | Read schema definitions.                                               |
| [`schema-writer`](/cloud/security/access/rbac/manage-rbac-roles#schema-writer)   | Create and update schemas.                                             |
| [`schema-manager`](/cloud/security/access/rbac/manage-rbac-roles#schema-manager) | Manage compatibility policies, plus everything `schema-writer` grants. |
| [`schema-owner`](/cloud/security/access/rbac/manage-rbac-roles#schema-owner)     | Full control, including deleting subjects.                             |

A producer that registers schemas needs `schema-writer`; a consumer that only resolves them needs
`schema-reader`. See
[Manage RBAC roles](/cloud/security/access/rbac/manage-rbac-roles#schema-registry) for how to bind a
role to a service account.

<Note title="Pulsar clusters using topic-level permissions">
  Some Pulsar clusters authorize Schema Registry access through a single ACL on the registry's backing
  topic instead of RBAC. See
  [Kafka Schema Registry on Pulsar clusters](/cloud/governance/sr/kafka-schema-registry#grant-access).
</Note>

## Verify the connection

List the schema formats the registry supports. A successful response confirms the URL and your
credentials:

```shell theme={null}
curl -u "any-user:$API_KEY" https://<schema-registry-url>/schemas/types
```

```json theme={null}
["AVRO","JSON","PROTOBUF"]
```

Then list the subjects registered on the cluster:

```shell theme={null}
curl -u "any-user:$API_KEY" https://<schema-registry-url>/subjects
```

<Note title="Note">
  `/subjects` returns only the subjects your credentials can read, rather than failing with a
  permission error. An empty array can mean the registry is empty *or* that your role doesn't cover any
  existing subject.
</Note>

## Next steps

<CardGroup cols={2}>
  <Card title="Confluent compatibility" icon="list-check" href="/kafka/governance/sr/reference/confluent-compatibility">
    Which endpoints work, which don't, and where behavior differs.
  </Card>

  <Card title="Schema ID validation" icon="shield-check" href="/kafka/governance/sr/manage/schema-id-validation">
    Make the broker reject records without a registered schema ID.
  </Card>
</CardGroup>