Validating JSON Messages with Schema Registry and JSON Schema

When Kafka topics carry JSON payloads, enforcing a structural contract at write time turns deserialization bugs into producer-side errors you can catch immediately, instead of latent corruption a consumer trips over hours later. Confluent Schema Registry — long associated with Avro and Protobuf — provides first-class JSON Schema support, so the same registration, validation, and compatibility machinery applies to JSON. This guide covers the end-to-end path: configuration, schema registration, producer and consumer serdes, the non-obvious evolution rules, and what to actually monitor in production.

JSON Schema Validation in the Schema Registry Ecosystem

Schema Registry is the central schema store within the broader Schema Registry & Connect Ecosystem. It stores and versions schemas and enforces compatibility on evolution. The validation itself happens client-side in the serializer: when you produce a record, the KafkaJsonSchemaSerializer resolves a schema for the subject (looking it up or registering it), validates the payload against it, prepends the 5-byte Confluent wire-format header (a magic byte plus the 4-byte schema ID), and writes the bytes. If the document does not satisfy the schema, the serializer throws a SerializationException and the record never reaches the broker.

This matters because validation at write time converts a class of latent failures into immediate, attributable ones. A malformed message that lands in a topic typically surfaces later as a consumer deserialization failure, often mid-incident and far from the producer that caused it. Schema Registry also enforces compatibility when schemas change — covered in Schema Evolution and Compatibility in Confluent Schema Registry — so understand your subject’s compatibility mode before modifying a production schema.

JSON Schema is a declarative vocabulary for validating JSON documents. Confluent’s implementation supports Drafts 4, 6, and 7 (via the Everit library) and Drafts 2019-09 and 2020-12 (via the json-sKema library), per the JSON Schema serdes documentation. You constrain fields with types, required properties, string patterns, numeric ranges, and nested-object rules, and the serializer enforces them.

Confirming JSON Schema Support

JSON Schema, Avro, and Protobuf providers are loaded by Schema Registry automatically — there is no config flag to “turn on” JSON, and no schema.providers or compatibility.json property in schema-registry.properties (those keys do not exist). The schema.providers mechanism exists only for registering additional, custom SchemaProvider plugins; you do not list the built-in formats there.

A standard schema-registry.properties only needs the store and listener wiring:

# Brokers backing the _schemas log (protocol-prefixed)
kafkastore.bootstrap.servers=PLAINTEXT://kafka-broker-1:9092,PLAINTEXT://kafka-broker-2:9092
kafkastore.topic=_schemas

# Listener for producer/consumer clients
listeners=http://0.0.0.0:8081

# Default compatibility level for new subjects (default is BACKWARD)
schema.compatibility.level=backward

Note schema.compatibility.level — the older avro.compatibility.level key is deprecated as of Confluent Platform 5.5. After starting Schema Registry, confirm it advertises the JSON type:

curl -s http://schema-registry:8081/schemas/types | jq .
# e.g. ["JSON","PROTOBUF","AVRO"]

GET /schemas/types returns the registered schema types as a JSON array; treat it as an unordered set. If "JSON" is missing, you are running a build that predates JSON Schema support — it shipped in Confluent Platform 5.5.

Defining and Registering a JSON Schema

A JSON Schema is the contract for a topic’s value. Consider a user-signup event:

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "$id": "https://example.com/schemas/user-signup.json",
  "title": "UserSignup",
  "type": "object",
  "properties": {
    "userId": {
      "type": "string",
      "pattern": "^[a-f0-9]{8}-[a-f0-9]{4}-4[a-f0-9]{3}-[89ab][a-f0-9]{3}-[a-f0-9]{12}$",
      "description": "UUID v4"
    },
    "email": { "type": "string", "format": "email" },
    "timestamp": { "type": "string", "format": "date-time" },
    "plan": { "type": "string", "enum": ["free", "pro", "enterprise"] }
  },
  "required": ["userId", "email", "timestamp"],
  "additionalProperties": false
}

additionalProperties: false is deliberate: it rejects unknown fields, which is what makes the contract strict and forces every new field through an explicit schema change. The trade-off is reduced evolution flexibility — a closed content model permits fewer compatible changes than an open one (covered under Schema Evolution below).

Register it against the subject user-signups-value (the default TopicNameStrategy derives <topic>-value for values, <topic>-key for keys). The schema must be passed as a JSON-encoded string, so the document is escaped inside the request body:

curl -X POST \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  --data '{"schemaType":"JSON","schema":"{\"$schema\":\"http://json-schema.org/draft-07/schema#\",\"$id\":\"https://example.com/schemas/user-signup.json\",\"title\":\"UserSignup\",\"type\":\"object\",\"properties\":{\"userId\":{\"type\":\"string\",\"pattern\":\"^[a-f0-9]{8}-[a-f0-9]{4}-4[a-f0-9]{3}-[89ab][a-f0-9]{3}-[a-f0-9]{12}$\"},\"email\":{\"type\":\"string\",\"format\":\"email\"},\"timestamp\":{\"type\":\"string\",\"format\":\"date-time\"},\"plan\":{\"type\":\"string\",\"enum\":[\"free\",\"pro\",\"enterprise\"]}},\"required\":[\"userId\",\"email\",\"timestamp\"],\"additionalProperties\":false}"}' \
  http://schema-registry:8081/subjects/user-signups-value/versions

A success returns the global schema id (e.g. {"id":1}). Confirm the registered version:

curl -s http://schema-registry:8081/subjects/user-signups-value/versions/latest | jq .

Producer-Side Validation (Java)

Configure the producer’s value serializer as KafkaJsonSchemaSerializer. In production, register schemas out-of-band (as above) and run clients with auto.register.schemas=false so application deploys cannot silently mutate the subject:

import io.confluent.kafka.serializers.json.KafkaJsonSchemaSerializer;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-broker-1:9092,kafka-broker-2:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaJsonSchemaSerializer.class);
props.put("schema.registry.url", "http://schema-registry:8081");
props.put("auto.register.schemas", "false");  // default true; disable in prod
props.put("use.latest.version", "true");       // default false; serialize against latest registered schema
props.put("json.fail.invalid.schema", "true"); // default false; reject payloads that violate the schema

KafkaProducer<String, JsonNode> producer = new KafkaProducer<>(props);

The load-bearing property is json.fail.invalid.schema. It defaults to false, meaning the serializer will encode and send a message that does not actually satisfy the schema — silent data corruption that only the schema, not your data, is “valid.” Set it to true so a violation raises a SerializationException at send() time. The serdes defaults are counterintuitive enough that they deserve a table:

Pairing auto.register.schemas=false with use.latest.version=true forces the serializer to use the schema already registered for the subject rather than inferring or registering one from the payload. normalize.schemas=false is a common source of avoidable version churn: re-registering a schema whose properties are merely reordered will mint a new version under the default, so enabling normalization is recommended.

Producer-Side Validation (Python)

The confluent-kafka package (PyPI name confluent-kafka, install with the [schemaregistry]/[json] extras) exposes the class as JSONSerializer in confluent_kafka.schema_registry.json_schema — there is no JSONSchemaSerializer. Its constructor takes the schema string, the registry client, and a to_dict callable that converts your object into a plain dict for serialization. Serializers are invoked inside produce() via a SerializationContext, so use SerializingProducer rather than the base Producer:

from confluent_kafka import SerializingProducer
from confluent_kafka.serialization import StringSerializer
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.json_schema import JSONSerializer

schema_registry_client = SchemaRegistryClient({'url': 'http://schema-registry:8081'})

# Use the schema already registered for the subject
latest = schema_registry_client.get_latest_version('user-signups-value')
schema_str = latest.schema.schema_str

def signup_to_dict(signup, ctx):
    return signup  # already a dict; map your object to a dict here

json_serializer = JSONSerializer(
    schema_str,
    schema_registry_client,
    to_dict=signup_to_dict,
    conf={'auto.register.schemas': False},
)

producer = SerializingProducer({
    'bootstrap.servers': 'kafka-broker-1:9092',
    'key.serializer': StringSerializer('utf_8'),
    'value.serializer': json_serializer,
})

producer.produce(topic='user-signups', key='k', value={'userId': '...', 'email': '...', 'timestamp': '...'})
producer.flush()

A payload that fails validation raises a ValueSerializationError from produce(), which the application can catch, log, and route to a dead-letter topic or alert. This keeps bad data out of the Schema Registry & Connect Ecosystem at the source.

Consumer-Side Deserialization

Producer-side validation is the primary defense, but consumers should still deserialize through the registry to guard against a producer that bypassed it or an unexpected compatibility gap. Configure KafkaJsonSchemaDeserializer:

import io.confluent.kafka.serializers.json.KafkaJsonSchemaDeserializer;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.util.Properties;

Properties consumerProps = new Properties();
consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-broker-1:9092");
consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "user-signups-consumer");
consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaJsonSchemaDeserializer.class);
consumerProps.put("schema.registry.url", "http://schema-registry:8081");
consumerProps.put("json.fail.invalid.schema", "true");

KafkaConsumer<String, JsonNode> consumer = new KafkaConsumer<>(consumerProps);

The deserializer reads the schema ID from the record’s wire-format header (not a Kafka message header), fetches that schema from the registry, and decodes the value. With json.fail.invalid.schema=true it also re-validates the decoded document. A failure throws on poll(), so handle the poison pill explicitly: record the offset, route the raw bytes aside, and alert rather than crash-loop. The default failure mode is an uncaught exception that the consumer reprocesses on every restart — a single bad record halting the partition indefinitely.

Schema Evolution and Compatibility

JSON Schema support lets you evolve a contract as long as the change satisfies the subject’s compatibility mode. Inspect the current setting before changing anything:

curl -s http://schema-registry:8081/config/user-signups-value | jq .

The response’s compatibilityLevel is one of BACKWARD, BACKWARD_TRANSITIVE, FORWARD, FORWARD_TRANSITIVE, FULL, FULL_TRANSITIVE, or NONE. BACKWARD (the default) means a consumer using the new schema can read data written with the previous one; FORWARD is the reverse; FULL requires both.

The non-obvious part of JSON Schema evolution is that compatibility depends on the content model, not just the compatibility level. Per the Confluent evolution docs, an open content model (additionalProperties: true) permits more changes than a closed one (additionalProperties: false). Adding or removing an optional field is backward-compatible under both, but the closed model rejects a range of changes the open model would accept:

So adding the optional referralSource below passes under BACKWARD for both models — but do not generalize that to type-widening or new required fields on a closed schema:

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "$id": "https://example.com/schemas/user-signup.json",
  "title": "UserSignup",
  "type": "object",
  "properties": {
    "userId": { "type": "string", "pattern": "^[a-f0-9]{8}-[a-f0-9]{4}-4[a-f0-9]{3}-[89ab][a-f0-9]{3}-[a-f0-9]{12}$" },
    "email": { "type": "string", "format": "email" },
    "timestamp": { "type": "string", "format": "date-time" },
    "plan": { "type": "string", "enum": ["free", "pro", "enterprise"] },
    "referralSource": { "type": "string" }
  },
  "required": ["userId", "email", "timestamp"],
  "additionalProperties": false
}

Because the closed model is restrictive, never guess — dry-run the check before registering with the compatibility endpoint, which returns {"is_compatible": true|false}:

curl -X POST \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  --data '{"schemaType":"JSON","schema":"...escaped schema..."}' \
  http://schema-registry:8081/compatibility/subjects/user-signups-value/versions/latest

If the registration (POST .../versions) violates the mode, Schema Registry rejects it with HTTP 409 — wire this check into CI so an incompatible change fails the build, as described in Schema Evolution and Compatibility in Confluent Schema Registry.

Monitoring and Operational Practices

Schema Registry reports its metrics over JMX, not a built-in Prometheus endpoint — there is no /metrics route on port 8081. To scrape it with Prometheus, run the Prometheus JMX Exporter as a Java agent via SCHEMA_REGISTRY_JMX_OPTS (or as a sidecar in Kubernetes) and point your scrape job at the exporter’s port:

scrape_configs:
  - job_name: 'schema-registry'
    static_configs:
      - targets: ['schema-registry:5556']  # JMX-exporter port, not 8081

Each failure mode has a distinct signal and a distinct cause — monitor the symptom, then read backward to the source:

Two operational rules follow from this:

• Enforce schema checks in CI, not at runtime. The Confluent Schema Registry Maven Plugin can run test-compatibility against the live registry as a build gate, so an incompatible change fails the build instead of surfacing as a 409 mid-deploy.
• Preserve poison pills verbatim. For records that still fail at the consumer, use a dead-letter topic that stores the raw bytes with a ByteArraySerializer/ByteArrayDeserializer, so the failing payload survives for inspection or replay without a second deserialization attempt that would just fail again.

Producer-side validation, consumer-side defense, compatibility enforcement in CI, and JMX-based monitoring together catch schema violations at the boundary instead of letting them propagate into an incident.