10.4.2 – Schema Registry REST API

Module: Section 10 – Kafka Ecosystem & Tools
Section: 10.4 – Schema Registry Operations
Goal: Master the Schema Registry REST API for operational tasks and automation

Introduction

Schema Registry exposes a comprehensive REST API for schema management, compatibility checking, and administrative operations.

Why This Module Matters: Understanding the REST API enables automation, CI/CD integration, and operational troubleshooting.

API Overview

Base URL: http://localhost:8081
Content-Type: application/vnd.schemaregistry.v1+json
Authentication: Basic Auth, OAuth, or mTLS (if configured)

Schema Registration

Register New Schema

POST /subjects/{subject}/versions

curl -X POST http://localhost:8081/subjects/users-value/versions \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -d '{
    "schema": "{\"type\":\"record\",\"name\":\"User\",\"fields\":[{\"name\":\"userId\",\"type\":\"long\"},{\"name\":\"username\",\"type\":\"string\"}]}"
  }'

Response:

{"id": 42}

Notes:

Returns global schema ID
Compatibility check performed automatically
Returns existing ID if schema already registered

Register Schema Without Compatibility Check

POST /subjects/{subject}/versions?normalize=true

curl -X POST "http://localhost:8081/subjects/users-value/versions?normalize=true" \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -d '{"schema": "..."}'

⚠️ Warning: Bypasses compatibility check. Use with caution!

Schema Retrieval

Get Schema by ID

GET /schemas/ids/{id}

Retrieve schema by global schema ID.

curl http://localhost:8081/schemas/ids/42

Response:

{
  "schema": "{\"type\":\"record\",\"name\":\"User\",\"fields\":[...]}"
}

Use Case: Consumers use this endpoint to deserialize messages.

Get Latest Schema for Subject

GET /subjects/{subject}/versions/latest

curl http://localhost:8081/subjects/users-value/versions/latest

Response:

{
  "subject": "users-value",
  "version": 3,
  "id": 42,
  "schema": "{\"type\":\"record\",\"name\":\"User\",\"fields\":[...]}"
}

Get Specific Version

GET /subjects/{subject}/versions/{version}

curl http://localhost:8081/subjects/users-value/versions/2

Response:

{
  "subject": "users-value",
  "version": 2,
  "id": 27,
  "schema": "{...}"
}

Subject Management

List All Subjects

GET /subjects

curl http://localhost:8081/subjects

Response:

["users-value", "orders-value", "payments-key"]

List Versions for Subject

GET /subjects/{subject}/versions

curl http://localhost:8081/subjects/users-value/versions

Response:

[1, 2, 3]

Delete Subject (Soft Delete)

DELETE /subjects/{subject}

curl -X DELETE http://localhost:8081/subjects/users-value

Response:

[1, 2, 3]

Effect: Soft delete (can be recovered).

Delete Subject (Permanent)

DELETE /subjects/{subject}?permanent=true

curl -X DELETE "http://localhost:8081/subjects/users-value?permanent=true"

⚠️ Warning: Irreversible! Consumers with messages referencing deleted schema IDs will fail.

Delete Specific Version

DELETE /subjects/{subject}/versions/{version}

curl -X DELETE http://localhost:8081/subjects/users-value/versions/2

Response:

Compatibility Checking

Check Compatibility

POST /compatibility/subjects/{subject}/versions/{version}

Test if a new schema is compatible with a specific version.

curl -X POST http://localhost:8081/compatibility/subjects/users-value/versions/latest \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -d '{
    "schema": "{\"type\":\"record\",\"name\":\"User\",\"fields\":[{\"name\":\"userId\",\"type\":\"long\"},{\"name\":\"username\",\"type\":\"string\"},{\"name\":\"email\",\"type\":[\"null\",\"string\"],\"default\":null}]}"
  }'

Response (Compatible):

{"is_compatible": true}

Response (Incompatible):

{
  "is_compatible": false,
  "messages": ["Reader field email not found in writer schema"]
}

Check Compatibility Against All Versions

POST /compatibility/subjects/{subject}/versions

curl -X POST http://localhost:8081/compatibility/subjects/users-value/versions \
  -d '{"schema": "..."}'

Use Case: TRANSITIVE compatibility modes check against all versions.

Configuration Management

Get Global Compatibility

GET /config

curl http://localhost:8081/config

Response:

{"compatibilityLevel": "BACKWARD"}

Set Global Compatibility

PUT /config

curl -X PUT http://localhost:8081/config \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -d '{"compatibility": "FULL"}'

Response:

{"compatibility": "FULL"}

Get Subject-Level Compatibility

GET /config/{subject}

curl http://localhost:8081/config/users-value

Response:

{"compatibilityLevel": "BACKWARD_TRANSITIVE"}

Set Subject-Level Compatibility

PUT /config/{subject}

curl -X PUT http://localhost:8081/config/users-value \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -d '{"compatibility": "BACKWARD_TRANSITIVE"}'

Delete Subject-Level Compatibility

DELETE /config/{subject}

Revert to global default.

curl -X DELETE http://localhost:8081/config/users-value

Mode Management

Get Mode

GET /mode

curl http://localhost:8081/mode

Response:

{"mode": "READWRITE"}

Modes:

READWRITE: Normal operation
READONLY: Block schema registration (read-only)
IMPORT: Import mode (for migrations)

Set Mode

PUT /mode

curl -X PUT http://localhost:8081/mode \
  -d '{"mode": "READONLY"}'

Use Case: Set READONLY during maintenance or schema freezes.

Advanced Operations

Get Schema Versions by Schema Content

POST /subjects/{subject}

Find versions matching specific schema.

curl -X POST http://localhost:8081/subjects/users-value \
  -d '{"schema": "..."}'

Response:

{
  "subject": "users-value",
  "version": 2,
  "id": 27,
  "schema": "{...}"
}

Get All Versions of a Schema ID

GET /schemas/ids/{id}/versions

Find all subjects/versions using a schema ID.

curl http://localhost:8081/schemas/ids/42/versions

Response:

[
  {"subject": "users-value", "version": 3},
  {"subject": "admins-value", "version": 1}
]

Automation Examples

CI/CD Schema Registration

#!/bin/bash

SCHEMA_FILE="user.avsc"
SUBJECT="users-value"
SCHEMA_REGISTRY="http://localhost:8081"

# Read schema from file
SCHEMA=$(cat $SCHEMA_FILE | jq -c .)

# Test compatibility
COMPAT_RESPONSE=$(curl -s -X POST "$SCHEMA_REGISTRY/compatibility/subjects/$SUBJECT/versions/latest" \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -d "{\"schema\": \"$SCHEMA\"}")

IS_COMPATIBLE=$(echo $COMPAT_RESPONSE | jq -r '.is_compatible')

if [ "$IS_COMPATIBLE" == "true" ]; then
  echo "✅ Schema is compatible. Registering..."
  
  # Register schema
  REGISTER_RESPONSE=$(curl -s -X POST "$SCHEMA_REGISTRY/subjects/$SUBJECT/versions" \
    -H "Content-Type: application/vnd.schemaregistry.v1+json" \
    -d "{\"schema\": \"$SCHEMA\"}")
  
  SCHEMA_ID=$(echo $REGISTER_RESPONSE | jq -r '.id')
  echo "✅ Schema registered with ID: $SCHEMA_ID"
else
  echo "❌ Schema is incompatible!"
  echo $COMPAT_RESPONSE | jq '.messages'
  exit 1
fi

Backup All Schemas

#!/bin/bash

SCHEMA_REGISTRY="http://localhost:8081"
BACKUP_DIR="schema-backup-$(date +%Y%m%d)"

mkdir -p $BACKUP_DIR

# Get all subjects
SUBJECTS=$(curl -s "$SCHEMA_REGISTRY/subjects" | jq -r '.[]')

for SUBJECT in $SUBJECTS; do
  echo "Backing up $SUBJECT..."
  
  # Get all versions
  VERSIONS=$(curl -s "$SCHEMA_REGISTRY/subjects/$SUBJECT/versions" | jq -r '.[]')
  
  for VERSION in $VERSIONS; do
    # Get schema
    SCHEMA=$(curl -s "$SCHEMA_REGISTRY/subjects/$SUBJECT/versions/$VERSION")
    
    # Save to file
    echo $SCHEMA | jq '.' > "$BACKUP_DIR/$SUBJECT-v$VERSION.json"
  done
done

echo "✅ Backup complete: $BACKUP_DIR"

Restore Schemas

#!/bin/bash

SCHEMA_REGISTRY="http://localhost:8081"
BACKUP_DIR="schema-backup-20250201"

for FILE in $BACKUP_DIR/*.json; do
  echo "Restoring $FILE..."
  
  SUBJECT=$(jq -r '.subject' $FILE)
  SCHEMA=$(jq -r '.schema' $FILE)
  
  # Register schema
  curl -X POST "$SCHEMA_REGISTRY/subjects/$SUBJECT/versions" \
    -H "Content-Type: application/vnd.schemaregistry.v1+json" \
    -d "{\"schema\": \"$SCHEMA\"}"
done

echo "✅ Restore complete"

Authentication

Basic Authentication

curl -u username:password \
  http://localhost:8081/subjects

Bearer Token (OAuth)

curl -H "Authorization: Bearer <token>" \
  http://localhost:8081/subjects

Client Certificate (mTLS)

curl --cert client.crt --key client.key --cacert ca.crt \
  https://localhost:8081/subjects

Error Codes

Code	Error	Description
40401	Subject Not Found	Subject doesn't exist
40402	Version Not Found	Version doesn't exist for subject
40403	Schema Not Found	Schema ID not found
42201	Invalid Schema	Schema syntax error
42202	Invalid Version	Version must be positive integer
409	Incompatible Schema	Schema violates compatibility rules
422	Unprocessable Entity	Standby instance (not primary)
500	Internal Server Error	Schema Registry internal error

CLI Tools for Schema Registry

There are 3 primary CLI tools to interact with Schema Registry:

1. `curl` (Direct REST API Access)

Most flexible for all operations:

# Register a schema
curl -X POST http://localhost:8081/subjects/users-value/versions \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -d '{"schema": "{\"type\":\"record\",\"name\":\"User\",\"fields\":[{\"name\":\"id\",\"type\":\"string\"}]}"}'

# List all subjects
curl http://localhost:8081/subjects

# Get latest schema version
curl http://localhost:8081/subjects/users-value/versions/latest

# Check compatibility
curl -X POST http://localhost:8081/compatibility/subjects/users-value/versions/latest \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -d '{"schema": "..."}'

2. Confluent CLI

Official tool with Schema Registry support (requires Confluent Platform):

# List subjects
confluent schema-registry subject list

# Get schema by ID
confluent schema-registry schema get --id 42

# Register a new schema
confluent schema-registry schema create \
  --subject users-value \
  --schema schema.avsc

# Delete a subject (soft delete)
confluent schema-registry subject delete --subject users-value

3. Kafka Avro Console Tools

For producing/consuming Avro data with Schema Registry:

Avro Producer:

kafka-avro-console-producer \
  --bootstrap-server localhost:9092 \
  --topic users \
  --property schema.registry.url=http://localhost:8081 \
  --property value.schema='{"type":"record","name":"User","fields":[{"name":"id","type":"string"}]}'

Avro Consumer:

kafka-avro-console-consumer \
  --bootstrap-server localhost:9092 \
  --topic users \
  --property schema.registry.url=http://localhost:8081 \
  --from-beginning

Use Cases:

Quick testing of Avro serialization
Debugging schema issues
Inspecting messages in Avro topics
Prototyping schema changes

CI/CD Integration

Common Use Cases

Schema Registry in CI/CD pipelines:

✅ Fail pipeline if schema breaks compatibility
✅ Auto-register new schemas from Avro definitions
✅ Validate schemas against existing versions
✅ Package schema registration with application builds

Example: GitHub Actions Workflow

name: Schema Validation

on: [pull_request]

jobs:
  validate-schema:
    runs-on: ubuntu-latest
    steps:
      - name: Checkout code
        uses: actions/checkout@v2
      
      - name: Check schema compatibility
        run: |
          SCHEMA=$(cat src/main/avro/user.avsc | jq -c .)
          RESPONSE=$(curl -s -X POST http://schema-registry:8081/compatibility/subjects/users-value/versions/latest \
            -H "Content-Type: application/vnd.schemaregistry.v1+json" \
            -d "{\"schema\": \"$SCHEMA\"}")
          
          IS_COMPATIBLE=$(echo $RESPONSE | jq -r '.is_compatible')
          
          if [ "$IS_COMPATIBLE" != "true" ]; then
            echo "❌ Schema is incompatible!"
            echo $RESPONSE | jq '.messages'
            exit 1
          fi
          
          echo "✅ Schema is compatible"
      
      - name: Register schema
        if: github.ref == 'refs/heads/main'
        run: |
          SCHEMA=$(cat src/main/avro/user.avsc | jq -c .)
          curl -X POST http://schema-registry:8081/subjects/users-value/versions \
            -H "Content-Type: application/vnd.schemaregistry.v1+json" \
            -d "{\"schema\": \"$SCHEMA\"}"

Example: Maven Build Integration

<plugin>
  <groupId>io.confluent</groupId>
  <artifactId>kafka-schema-registry-maven-plugin</artifactId>
  <version>7.5.0</version>
  <configuration>
    <schemaRegistryUrls>
      <param>http://localhost:8081</param>
    </schemaRegistryUrls>
    <subjects>
      <users-value>src/main/avro/user.avsc</users-value>
    </subjects>
  </configuration>
  <executions>
    <execution>
      <goals>
        <goal>test-compatibility</goal>
        <goal>register</goal>
      </goals>
    </execution>
  </executions>
</plugin>

Best Practices for CI/CD

Practice	Why It Matters
Test compatibility before merge	Catch breaking changes early
Auto-register in deployment	Ensure schemas exist before app starts
Version control schemas	Track schema evolution in Git
Separate dev/prod registries	Test schema changes safely
Document breaking changes	Help consumers plan updates

Example Flow:

Developer commits schema change
  ↓
CI pipeline runs
  ↓
Test compatibility against dev Schema Registry
  ↓
If compatible: Register in dev → Deploy app → Test
  ↓
If tests pass: Merge to main
  ↓
Production pipeline: Register in prod Schema Registry → Deploy app

Key Takeaways

Base URL: http://localhost:8081 (default Schema Registry endpoint).
Register Schema: POST /subjects/{subject}/versions.
Get Schema: GET /schemas/ids/{id} (consumers use this).
Check Compatibility: POST /compatibility/subjects/{subject}/versions/latest.
List Subjects: GET /subjects.
Delete Subject: DELETE /subjects/{subject} (soft delete by default).
Set Compatibility: PUT /config/{subject}.
Mode Management: PUT /mode (READWRITE, READONLY, IMPORT).
CI/CD Integration: Test compatibility before registering schema.
Backup/Restore: Automate schema backup using REST API.

What's Next?

Next Module: 10.4.3 – Monitoring Schema Registry

What We Covered:

✅ Schema Registry REST API endpoints
✅ Registration, retrieval, compatibility
✅ CI/CD automation
✅ Backup/restore

What's Coming:

❌ Monitoring metrics and health checks
❌ Troubleshooting schema issues

References:

Subject naming → 10.4.1: Subject Naming Strategies
Compatibility → 10.3.2: Compatibility Types
Troubleshooting → 10.4.4: Troubleshooting Schema Issues

10.4.2 – Schema Registry REST API

Introduction

API Overview

Schema Registration

Register New Schema

Register Schema Without Compatibility Check

Schema Retrieval

Get Schema by ID

Get Latest Schema for Subject

Get Specific Version

Subject Management

List All Subjects

List Versions for Subject

Delete Subject (Soft Delete)

Delete Subject (Permanent)

Delete Specific Version

Compatibility Checking

Check Compatibility

Check Compatibility Against All Versions

Configuration Management

Get Global Compatibility

Set Global Compatibility

Get Subject-Level Compatibility

Set Subject-Level Compatibility

Delete Subject-Level Compatibility

Mode Management

Get Mode

Set Mode

Advanced Operations

Get Schema Versions by Schema Content

Get All Versions of a Schema ID

Automation Examples

CI/CD Schema Registration

Backup All Schemas

Restore Schemas

Authentication

Basic Authentication

Bearer Token (OAuth)

Client Certificate (mTLS)

Error Codes

CLI Tools for Schema Registry

1. curl (Direct REST API Access)

2. Confluent CLI

3. Kafka Avro Console Tools

CI/CD Integration

Common Use Cases

Example: GitHub Actions Workflow

Example: Maven Build Integration

Best Practices for CI/CD

Key Takeaways

What's Next?

1. `curl` (Direct REST API Access)