How to Monitor Confluent Schema Registry with Vigilmon

Confluent Schema Registry sits at the heart of every event-driven architecture that uses Avro, Protobuf, or JSON Schema serialization. When Schema Registry becomes unavailable, Kafka producers that use the Confluent serializer fail immediately — they can't look up schema IDs to serialize messages. Consumers fail too, unable to deserialize existing messages. The entire event pipeline grinds to a halt, and the error manifests as a cascade of serialization failures in dozens of services simultaneously.

Vigilmon gives you external visibility into Schema Registry availability, compatibility check latency, and replication health — before a Schema Registry outage takes down your entire Kafka ecosystem.

Why Schema Registry Needs External Monitoring

Schema Registry exposes a REST API on port 8081 with standard health endpoints. But you need more than a basic liveness check:

• REST API availability — Schema Registry's / endpoint returns its version; a non-200 means all producers and consumers using it will fail
• Schema registration failure detection — registration attempts that fail silently indicate write unavailability while reads still work
• Compatibility check latency — slow compatibility checks indicate leader pressure or replication lag, which can cause producer timeouts before they become errors
• Replication health — Schema Registry stores schemas in an internal Kafka topic (_schemas); replication issues cause split-brain where different instances return different schemas
• Subject and version staleness — a Schema Registry that hasn't seen a new registration in hours during an active system is likely disconnected from the Kafka cluster

Step 1: Use Schema Registry's Built-In Health Endpoints

Confluent Schema Registry exposes HTTP endpoints you can probe directly:

# Basic availability — returns Schema Registry version info
curl -i http://localhost:8081/

# Full cluster status including leader election
curl -s http://localhost:8081/v1/metadata/id

# List all subjects — proves read access to _schemas topic
curl -s http://localhost:8081/subjects

# Schema Registry mode (READWRITE / READONLY / IMPORT)
curl -s http://localhost:8081/mode

Example responses:

// GET / — healthy
{"kafkaStore":{"status":"STARTED","stateChangeEpoch":12},"leaderElection":{"leaderEpoch":5,"leaderIdentity":"schema-registry-1:8081"}}

// GET /mode — healthy
{"mode":"READWRITE"}

// GET /mode — degraded (read-only means writes will fail)
{"mode":"READONLY"}

Step 2: Build a Comprehensive Health Endpoint

The built-in endpoints tell you if Schema Registry is up, but not if schema registration actually works or if compatibility checks are slow. Build a thin health sidecar that validates both.

# schema_registry_health.py
import os
import time
import requests
from flask import Flask, jsonify

app = Flask(__name__)

SR_URL = os.environ.get('SCHEMA_REGISTRY_URL', 'http://localhost:8081')
TEST_SUBJECT = os.environ.get('HEALTH_CHECK_SUBJECT', '_health_check_subject')
MAX_COMPAT_LATENCY_MS = int(os.environ.get('MAX_COMPAT_LATENCY_MS', '2000'))

# A minimal Avro schema for the health check registration test
TEST_SCHEMA = '{"type":"record","name":"HealthCheck","fields":[{"name":"ts","type":"long"}]}'

@app.route('/health/schema-registry')
def health():
    issues = []

    # 1. Basic availability
    try:
        resp = requests.get(f'{SR_URL}/', timeout=5)
        if resp.status_code != 200:
            return jsonify({'status': 'down', 'reason': 'api_unavailable'}), 503
        info = resp.json()
    except Exception as e:
        return jsonify({'status': 'down', 'error': str(e)}), 503

    # 2. Mode check — READONLY means write operations will fail
    try:
        mode_resp = requests.get(f'{SR_URL}/mode', timeout=5)
        mode = mode_resp.json().get('mode', 'UNKNOWN')
        if mode != 'READWRITE':
            issues.append({'type': 'readonly_mode', 'mode': mode})
    except Exception as e:
        issues.append({'type': 'mode_check_failed', 'error': str(e)})

    # 3. Schema registration write test
    try:
        register_resp = requests.post(
            f'{SR_URL}/subjects/{TEST_SUBJECT}/versions',
            json={'schema': TEST_SCHEMA},
            headers={'Content-Type': 'application/vnd.schemaregistry.v1+json'},
            timeout=5,
        )
        if register_resp.status_code not in (200, 409):  # 409 = schema already registered
            issues.append({
                'type': 'registration_failed',
                'status': register_resp.status_code,
                'detail': register_resp.text[:200],
            })
    except Exception as e:
        issues.append({'type': 'registration_error', 'error': str(e)})

    # 4. Compatibility check latency
    try:
        start = time.monotonic()
        compat_resp = requests.post(
            f'{SR_URL}/compatibility/subjects/{TEST_SUBJECT}/versions/latest',
            json={'schema': TEST_SCHEMA},
            headers={'Content-Type': 'application/vnd.schemaregistry.v1+json'},
            timeout=10,
        )
        latency_ms = int((time.monotonic() - start) * 1000)
        if latency_ms > MAX_COMPAT_LATENCY_MS:
            issues.append({
                'type': 'compatibility_check_slow',
                'latency_ms': latency_ms,
                'threshold_ms': MAX_COMPAT_LATENCY_MS,
            })
    except Exception as e:
        issues.append({'type': 'compatibility_check_error', 'error': str(e)})

    if issues:
        return jsonify({
            'status': 'degraded',
            'issues': issues,
            'schema_registry_version': info.get('version', 'unknown'),
        }), 503

    return jsonify({
        'status': 'ok',
        'schema_registry_version': info.get('version', 'unknown'),
        'mode': 'READWRITE',
    }), 200

if __name__ == '__main__':
    app.run(port=8092)

Deploy this sidecar alongside your Schema Registry instance. It validates:

1. REST API availability
2. Write mode (not READONLY)
3. Schema registration (an actual write to the _schemas topic)
4. Compatibility check latency (a good proxy for leader/replication pressure)

Step 3: Configure Vigilmon Monitors

Monitor 1: Schema Registry Availability

1. Log in to vigilmon.online → Monitors → New Monitor
2. Choose HTTP / HTTPS
3. URL: https://your-schema-registry.example.com:8081/
4. Check interval: 1 minute
5. Expected:
   • Status code: 200
   • Response time threshold: 5000ms
6. Alert channel: Slack + PagerDuty (this is P1 — Schema Registry down = entire Kafka ecosystem down)
7. Save

Monitor 2: Comprehensive Health Check

Add a second monitor for the health sidecar that catches degraded states before they become outages:

• URL: https://your-schema-registry-sidecar.example.com:8092/health/schema-registry
• Expected: 200, body contains "status":"ok"
• Interval: 2 minutes
• Alert channel: Slack (P2 — degraded before it becomes P1)

Monitor 3: Schema Registry Mode

Probe the mode endpoint directly to catch READONLY transitions (which prevent all writes):

• URL: https://your-schema-registry.example.com:8081/mode
• Expected: 200, body contains READWRITE
• Interval: 1 minute
• Alert channel: Slack + PagerDuty (mode flip to READONLY is effectively a write outage)

Step 4: Multi-Instance Replication Monitoring

If you run multiple Schema Registry instances for high availability, replication failures can cause instances to diverge — different instances returning different schema IDs for the same subject. Add monitors for each instance:

# For each Schema Registry replica, probe its leader election state
# A non-leader should redirect reads to the leader — if it doesn't, replication is broken
curl -s http://schema-registry-2:8081/v1/metadata/id
curl -s http://schema-registry-2:8081/subjects  # Should match schema-registry-1

Build a cross-instance consistency check:

@app.route('/health/schema-registry/replication')
def replication_health():
    instances = os.environ.get('SR_INSTANCES', 'http://sr-1:8081,http://sr-2:8081').split(',')
    subject_counts = {}

    for instance in instances:
        try:
            resp = requests.get(f'{instance.strip()}/subjects', timeout=5)
            subject_counts[instance] = len(resp.json()) if resp.status_code == 200 else -1
        except Exception:
            subject_counts[instance] = -1

    counts = list(subject_counts.values())
    if len(set(c for c in counts if c >= 0)) > 1:
        return jsonify({
            'status': 'degraded',
            'reason': 'replication_divergence',
            'subject_counts': subject_counts,
        }), 503

    if any(c < 0 for c in counts):
        return jsonify({
            'status': 'degraded',
            'reason': 'instance_unreachable',
            'subject_counts': subject_counts,
        }), 503

    return jsonify({'status': 'ok', 'subject_counts': subject_counts}), 200

Step 5: Alert Routing

| Monitor | Alert Channel | Priority | |---|---|---| | Schema Registry / availability | Slack + PagerDuty | P1 | | Mode endpoint (READWRITE check) | Slack + PagerDuty | P1 | | Health sidecar (registration + latency) | Slack | P2 | | Replication consistency | Slack | P2 |

Set response time thresholds on compatibility checks — latency above 2 seconds typically precedes leader election pressure, giving you warning before producers start timing out.

Summary

Schema Registry is a single point of failure for Kafka-based event architectures. External monitoring catches availability failures, READONLY mode transitions, and replication divergence before they cascade to producers and consumers.

| Monitor Type | What It Covers | |---|---| | HTTP monitor on / | Basic REST API availability | | HTTP monitor on /mode | READWRITE vs READONLY state | | HTTP monitor on health sidecar | Schema registration writes, compatibility latency | | HTTP monitor on replication check | Multi-instance divergence detection |

Get started free at vigilmon.online — your first Schema Registry monitor is running in under two minutes.