tutorial

How to Monitor Confluent Schema Registry with Vigilmon

Confluent Schema Registry manages Avro, JSON Schema, and Protobuf schemas for Kafka. Learn how to monitor registry API health, schema compatibility check latency, subject count, and storage backend health with Vigilmon.

Confluent Schema Registry is the schema management layer for Apache Kafka ecosystems. It stores and evolves Avro, JSON Schema, and Protobuf schemas that Kafka producers and consumers agree on at runtime. When Schema Registry is unavailable or degraded, Kafka producers can't serialize new messages, consumers can't deserialize existing ones, and your entire data pipeline grinds to a halt — even though Kafka itself is perfectly healthy.

This tutorial shows you how to monitor Confluent Schema Registry's REST API health, schema compatibility check latency, subject count, and storage backend health with Vigilmon HTTP monitors. You'll have end-to-end Schema Registry observability configured in under ten minutes.

Why Schema Registry Monitoring Is Non-Negotiable

Schema Registry sits on the critical path of every Kafka producer. If it goes down:

  • Producers fail immediately — serializers calling /subjects/{subject}/versions return Connection refused and the producer throws an exception
  • Consumers fail on new schema versions — io.confluent.kafka.serializers.KafkaAvroDeserializer fetches schemas at deserialization time; a registry outage means any message with an unresolved schema ID causes a SerializationException
  • Schema evolution breaks silently — if the registry returns 200 OK but with stale data (Kafka-backed storage lagging), producers and consumers may be operating on different schema versions

Vigilmon catches these conditions before your producers surface them.

Step 1: Verify Schema Registry's Built-In Endpoints

Schema Registry exposes a REST API with several built-in health signals:

# Health check — returns registered schemas overview
curl -s http://localhost:8081/ | jq .

# List all subjects
curl -s http://localhost:8081/subjects | jq .

# Schema compatibility mode
curl -s http://localhost:8081/config | jq .
# Expected: {"compatibilityLevel":"BACKWARD"}

# Test a specific subject's latest schema
curl -s http://localhost:8081/subjects/my-topic-value/versions/latest | jq '{id:.id, schemaType:.schemaType}'

Check whether Schema Registry's Kafka storage backend is healthy:

# Check internal topics exist
curl -s http://localhost:8081/v1/metadata/id | jq .
# Returns: {"scope":{"path":[],"clusters":{"kafka-cluster":"<cluster-id>"}},"id":"<schema-registry-id>"}

Step 2: Build a Schema Registry Health Probe

Create a health probe that aggregates the key signals into a single Vigilmon-readable endpoint:

# schema_registry_probe.py
import os
import time
import json
import requests
from http.server import HTTPServer, BaseHTTPRequestHandler

SR_URL = os.environ.get('SCHEMA_REGISTRY_URL', 'http://localhost:8081')
PROBE_PORT = int(os.environ.get('PROBE_PORT', '8082'))
COMPAT_CHECK_TIMEOUT = int(os.environ.get('COMPAT_CHECK_TIMEOUT_MS', '3000')) / 1000
SUBJECT_COUNT_ALERT = int(os.environ.get('SUBJECT_COUNT_ALERT', '0'))  # 0 = disabled

AUTH = None
sr_user = os.environ.get('SCHEMA_REGISTRY_USER')
sr_pass = os.environ.get('SCHEMA_REGISTRY_PASS')
if sr_user and sr_pass:
    AUTH = (sr_user, sr_pass)

def sr_get(path, timeout=5):
    return requests.get(f'{SR_URL}{path}', auth=AUTH, timeout=timeout)

def check_registry_up():
    """Verify the registry root endpoint responds."""
    try:
        r = sr_get('/')
        if r.status_code == 200:
            return True, r.json()
        return False, f'root_status_{r.status_code}'
    except Exception as e:
        return False, str(e)

def check_subjects():
    """List all subjects and return the count."""
    try:
        r = sr_get('/subjects')
        if r.status_code == 200:
            subjects = r.json()
            return True, len(subjects), subjects
        return False, 0, f'subjects_status_{r.status_code}'
    except Exception as e:
        return False, 0, str(e)

def check_compat_latency(subject=None):
    """Measure compatibility check round-trip time."""
    if not subject:
        # Find any subject to test against
        try:
            r = sr_get('/subjects')
            subjects = r.json() if r.status_code == 200 else []
            if not subjects:
                return None, 0.0, 'no_subjects_to_test'
            subject = subjects[0]
        except Exception:
            return None, 0.0, 'subject_list_failed'

    # Fetch latest schema for the subject
    try:
        r = sr_get(f'/subjects/{subject}/versions/latest')
        if r.status_code != 200:
            return subject, 0.0, f'latest_schema_{r.status_code}'
        schema_data = r.json()
        schema_str = schema_data.get('schema', '{}')
        schema_type = schema_data.get('schemaType', 'AVRO')
    except Exception as e:
        return subject, 0.0, str(e)

    # Test compatibility check latency
    payload = {'schema': schema_str, 'schemaType': schema_type}
    try:
        start = time.monotonic()
        r = requests.post(
            f'{SR_URL}/compatibility/subjects/{subject}/versions/latest',
            json=payload,
            auth=AUTH,
            timeout=COMPAT_CHECK_TIMEOUT,
        )
        latency_ms = (time.monotonic() - start) * 1000
        if r.status_code in (200, 404):  # 404 = no previous version
            return subject, round(latency_ms, 1), None
        return subject, round(latency_ms, 1), f'compat_api_{r.status_code}'
    except requests.Timeout:
        return subject, COMPAT_CHECK_TIMEOUT * 1000, 'compat_check_timeout'
    except Exception as e:
        return subject, 0.0, str(e)

def check_global_config():
    """Verify the global compatibility configuration is readable."""
    try:
        r = sr_get('/config')
        if r.status_code == 200:
            return True, r.json().get('compatibilityLevel', 'UNKNOWN')
        return False, f'config_status_{r.status_code}'
    except Exception as e:
        return False, str(e)

class ProbeHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path == '/health/schema-registry':
            self._check_all()
        elif self.path == '/health/schema-registry/subjects':
            self._check_subjects()
        elif self.path.startswith('/health/schema-registry/compat'):
            self._check_compat()
        elif self.path == '/health/schema-registry/config':
            self._check_config()
        else:
            self.send_response(404)
            self.end_headers()

    def _respond(self, code, body):
        self.send_response(code)
        self.send_header('Content-Type', 'application/json')
        self.end_headers()
        self.wfile.write(json.dumps(body).encode())

    def _check_all(self):
        issues = []
        up, info = check_registry_up()
        if not up:
            return self._respond(503, {'status': 'down', 'reason': info})

        subjects_ok, count, _ = check_subjects()
        if not subjects_ok:
            issues.append(f'subjects_api_failed')

        config_ok, compat_level = check_global_config()
        if not config_ok:
            issues.append('config_api_failed')

        subject, lat_ms, compat_err = check_compat_latency()
        compat_threshold = float(os.environ.get('COMPAT_LATENCY_THRESHOLD_MS', '2000'))
        if compat_err:
            issues.append(f'compat_check_error: {compat_err}')
        elif lat_ms > compat_threshold:
            issues.append(f'compat_check_slow_{lat_ms}ms_threshold_{compat_threshold}ms')

        if SUBJECT_COUNT_ALERT > 0 and count < SUBJECT_COUNT_ALERT:
            issues.append(f'subject_count_{count}_below_expected_{SUBJECT_COUNT_ALERT}')

        if issues:
            return self._respond(503, {'status': 'degraded', 'issues': issues,
                                       'subject_count': count})
        self._respond(200, {
            'status': 'ok',
            'subject_count': count,
            'compatibility_level': compat_level if config_ok else 'unknown',
            'compat_check_latency_ms': lat_ms,
            'tested_subject': subject,
        })

    def _check_subjects(self):
        ok, count, result = check_subjects()
        if not ok:
            return self._respond(503, {'status': 'down', 'reason': result})
        self._respond(200, {'status': 'ok', 'subject_count': count})

    def _check_compat(self):
        from urllib.parse import urlparse, parse_qs
        params = parse_qs(urlparse(self.path).query)
        subject = (params.get('subject') or [None])[0]
        subj, lat_ms, err = check_compat_latency(subject)
        threshold = float(os.environ.get('COMPAT_LATENCY_THRESHOLD_MS', '2000'))
        if err:
            return self._respond(503, {'status': 'down', 'reason': err, 'subject': subj})
        if lat_ms > threshold:
            return self._respond(503, {
                'status': 'degraded',
                'reason': 'compat_check_slow',
                'latency_ms': lat_ms,
                'threshold_ms': threshold,
                'subject': subj,
            })
        self._respond(200, {'status': 'ok', 'latency_ms': lat_ms, 'subject': subj})

    def _check_config(self):
        ok, result = check_global_config()
        if not ok:
            return self._respond(503, {'status': 'down', 'reason': result})
        self._respond(200, {'status': 'ok', 'compatibility_level': result})

    def log_message(self, *args):
        pass

if __name__ == '__main__':
    print(f'Schema Registry probe listening on :{PROBE_PORT}')
    HTTPServer(('0.0.0.0', PROBE_PORT), ProbeHandler).serve_forever()

Run it:

pip install requests
SCHEMA_REGISTRY_URL=http://localhost:8081 \
PROBE_PORT=8082 \
COMPAT_LATENCY_THRESHOLD_MS=2000 \
python schema_registry_probe.py

For Confluent Cloud with authentication:

SCHEMA_REGISTRY_URL=https://psrc-xxxxx.us-east-2.aws.confluent.cloud \
SCHEMA_REGISTRY_USER=your-api-key \
SCHEMA_REGISTRY_PASS=your-api-secret \
COMPAT_LATENCY_THRESHOLD_MS=3000 \
python schema_registry_probe.py

Step 3: Monitor Storage Backend Health

Schema Registry stores schemas in a Kafka internal topic (_schemas by default). If the Kafka cluster is degraded, Schema Registry can serve stale data from its in-memory cache while writes fail silently.

Add a write-path check:

def check_write_path(test_subject='__vigilmon-probe'):
    """Attempt to register a test schema version to verify the write path."""
    test_schema = json.dumps({
        "type": "record",
        "name": "VigilmonProbe",
        "namespace": "com.vigilmon",
        "fields": [{"name": "ts", "type": "long"}]
    })
    try:
        # Register the schema
        r = requests.post(
            f'{SR_URL}/subjects/{test_subject}/versions',
            json={'schema': test_schema, 'schemaType': 'AVRO'},
            auth=AUTH,
            timeout=5,
        )
        if r.status_code not in (200, 409):  # 409 = schema already exists
            return False, f'write_failed_{r.status_code}: {r.text[:200]}'
        schema_id = r.json().get('id')
        return True, schema_id
    except Exception as e:
        return False, str(e)

Expose this at /health/schema-registry/write:

elif self.path == '/health/schema-registry/write':
    ok, result = check_write_path()
    if not ok:
        return self._respond(503, {'status': 'down', 'reason': result,
                                   'detail': 'Kafka _schemas topic write path failed'})
    self._respond(200, {'status': 'ok', 'test_schema_id': result})

Step 4: Configure Vigilmon Monitors

  1. Log in to vigilmon.online and go to Monitors → New Monitor
  2. Choose HTTP / HTTPS
  3. Set the URL to your probe: https://your-host.example.com/health/schema-registry
  4. Set the check interval to 1 minute
  5. Under Expected response, configure:
    • Status code: 200
    • Response body contains: "status":"ok"
    • Response time threshold: 5000ms
  6. Under Alert channels, assign your Slack or PagerDuty integration

Full monitor table:

| Monitor URL | Purpose | Interval | Alert | |---|---|---|---| | /health/schema-registry | Full health: up, subjects, compat latency | 1 min | P1 – PagerDuty | | http://sr:8081/subjects | Raw subjects API availability | 1 min | P1 – PagerDuty | | /health/schema-registry/compat | Compatibility check round-trip latency | 2 min | P2 – Slack | | /health/schema-registry/write | Storage backend write path | 5 min | P1 – PagerDuty | | /health/schema-registry/config | Global compatibility config readable | 5 min | P2 – Slack |

Step 5: Latency Threshold Tuning

Compatibility check latency is the most actionable performance signal. Use these baseline thresholds:

| Deployment Type | P50 Expected | Alert Threshold | |---|---|---| | Local / Docker Compose | < 50ms | 500ms | | Self-hosted on-prem | < 100ms | 1000ms | | Confluent Cloud (same region) | < 200ms | 2000ms | | Confluent Cloud (cross-region) | < 500ms | 3000ms |

Monitor compatibility check latency trends over time in Vigilmon's response time charts. A gradual increase (50ms → 800ms over a week) indicates Kafka topic compaction lag or Schema Registry memory pressure — both resolvable before they become incidents.

Step 6: Alert Routing

| Monitor | Priority | Channel | Reason | |---|---|---|---| | Full health | P1 | PagerDuty | Producers and consumers both fail | | Write path | P1 | PagerDuty | New schema versions can't be registered | | Raw subjects API | P1 | PagerDuty | Registry returning errors | | Compat latency > threshold | P2 | Slack | Schema evolution is slow | | Config readable | P2 | Slack | Compatibility settings may be misconfigured |

Summary

Schema Registry is the silent dependency that makes Kafka serialization work. When it fails, your entire Kafka data pipeline stalls — but Kafka dashboards show green. The five Vigilmon monitors in this tutorial give you Registry-specific observability covering the REST API, the Kafka storage backend write path, compatibility check performance, and subject count integrity.

| Monitor | What It Catches | |---|---| | /health/schema-registry | Registry down, Kafka connectivity lost, slow compat checks | | Raw /subjects API | REST layer serving errors | | Write path probe | Kafka _schemas topic write failure | | Compat latency | Schema Registry memory pressure, Kafka lag | | Global config | Compatibility level misconfiguration |

Get started free at vigilmon.online — Schema Registry monitoring set up in minutes.

Monitor your app with Vigilmon

Free plan — 5 monitors, no credit card required. Up and running in 60 seconds.

Start free →
← Back to all articles