A JSONSchema to describe the shape of Apache Kafka's protocol message schemas.

kafka-meta.schema.json

{
    "$schema": "https://json-schema.org/draft/2020-12/schema",
    "type": "object",
    "additionalProperties": false,
    "oneOf": [
        {
            "properties": {
                "apiKey": {"type": "number"},
                "type": {
                    "type": "string",
                    "enum": ["request", "response"]
                }
            },
            "required": ["apiKey"]
        },
        {
            "properties": {
                "type": {
                    "type": "string",
                    "enum": ["data", "header"]
                }
            }
        }
    ],
    "properties": {
        "apiKey": {"type": "number"},
        "listeners": {
            "type": "array",
            "items": {"type": "string"}
        },
        "type": {
            "type": "string",
            "enum": ["request", "response", "data", "header"]
        },
        "name": {
            "type": "string",
            "minimum": 1
        },
        "validVersions": {"$ref": "#/$defs/versions"},
        "flexibleVersions": {"$ref": "#/$defs/versions"},
        "latestVersionUnstable": {"type": "boolean"},
        "fields": {
            "type": "array",
            "items": {"$ref":  "#/$defs/field"}
        },
        "commonStructs": {
            "type": "array",
            "items": {
                "type": "object",
                "additionalProperties": false,
                "properties": {
                    "name": {"type": "string"},
                    "versions": {"$ref": "#/$defs/versions"},
                    "fields": {
                        "type": "array",
                        "items": {"$ref": "#/$defs/field"}
                    }
                },
                "required": ["name", "versions", "fields"]
            }
        }
    },
    "required": ["type", "name", "validVersions", "flexibleVersions", "fields"],
    "$defs": {
        "field": {
            "type": "object",
            "additionalProperties": false,
            "properties": {
                "name": {"type": "string"},
                "about": {"type": "string"},
                "type": {
                    "type": "string",
                    "oneOf": [
                        {
                            "enum": [
                                "int8",
                                "int16",
                                "int32",
                                "int64",
                                "uint16",
                                "float64",
                                "string",
                                "bytes",
                                "records",
                                "bool",
                                "uuid",
                                "[]int8",
                                "[]int16",
                                "[]int32",
                                "[]int64",
                                "[]uint16",
                                "[]float64",
                                "[]string",
                                "[]bytes",
                                "[]records",
                                "[]bool",
                                "[]uuid"
                            ]
                        },
                        {"pattern": "^(?:\\[\\])?[A-Z]"}
                    ]
                },
                "entityType": {
                    "type": "string",
                    "enum": [
                        "topicName",
                        "brokerId",
                        "producerId",
                        "transactionalId",
                        "groupId"
                    ]
                },
                "default": {
                    "oneOf": [
                        {"type": "string"},
                        {"type": "number"},
                        {"type": "boolean"}
                    ]
                },
                "versions": {"$ref": "#/$defs/versions"},
                "nullableVersions": {"$ref": "#/$defs/versions"},
                "taggedVersions": {"$ref": "#/$defs/versions"},
                "tag": {"type": "number"},
                "flexibleVersions": {"$ref": "#/$defs/versions"},
                "ignorable": {"type": "boolean"},
                "fields": {
                    "type": "array",
                    "items": {"$ref": "#/$defs/field"}
                },
                "mapKey": {"type": "boolean"},
                "zeroCopy": {"type": "boolean"}
            },
            "required": ["name", "type", "versions"],
            "dependentRequired": {
                "tag": ["taggedVersions"],
                "taggedVersions": ["tag"]
            }
        },
        "versions": {
            "type": "string",
            "minimum": 1,
            "oneOf": [
                {"pattern": "^[0-9]+(?:\\+|-[0-9]+|)$"},
                {"enum": ["none"]}
            ]
        }
    }
}

requirements.txt

#
# This file is autogenerated by pip-compile with Python 3.12
# by the following command:
#
#    pip-compile --strip-extras
#
attrs==23.2.0
    # via
    #   jsonschema
    #   referencing
jsonschema==4.21.1
    # via -r requirements.in
jsonschema-specifications==2023.12.1
    # via jsonschema
referencing==0.33.0
    # via
    #   jsonschema
    #   jsonschema-specifications
rpds-py==0.17.1
    # via
    #   jsonschema
    #   referencing

verify.py

from typing import Iterator

from jsonschema import validate
import pathlib
import json


def read_strip_comments(path: pathlib.Path) -> bytes:
    def read_stripped() -> Iterator[bytes]:
        with path.open("rb") as fd:
            for line in fd.readlines():
                if line.lstrip(b" ").startswith(b"//"):
                    continue
                yield line

    return b"".join(read_stripped())


meta_schema_path = pathlib.Path("meta-schema.json")
meta_schema = json.loads(meta_schema_path.read_bytes())


schema_glob = pathlib.Path("schema/3.6.0/").glob("*.json")

for schema_path in schema_glob:
    print(schema_path)
    schema = json.loads(read_strip_comments(schema_path))
    validate(
        instance=schema,
        schema=meta_schema,
    )