Schema Registry and Multi-Protocol Serializer Hub

This document describes the Schema Registry and Multi-Protocol Serializer Hub in rekuiper.

Overview

The Schema Registry manages message schemas and data contracts for edge streaming. It provides contract management and parsing with zero external dependencies.

The registry supports four serialization formats:

1. Google Protocol Buffers v3 (protobuf): Strongly-typed binary contracts with message type extraction.
2. Custom Binary Codecs (custom): User-defined binary layouts and register specifications (e.g. Modbus RTU).
3. JSON Schema (json): JSON Schema Draft-07 structured object validation contracts.
4. Delimited / CSV (delimited): Field delimiter specifications for tabular and time-series streams.

---

Architecture Flow

The diagram below shows how payloads move from edge sensors to the Schema Registry and through the stream processing engine:

---

Key Capabilities

1. Proto3 AST Parser and Symbol Extractor

The Schema Registry contains an in-memory client-side AST parser for Protocol Buffers version 3:

• Detects package declarations (e.g. package rekuiper.industrial;).
• Extracts message identifiers and their typed field definitions with tag numbers.
• Identifies enumeration types and numeric values.
• Validates syntax and braces without compiling external binaries.

2. Live Stream Designer Integration

When configuring a Stream or Table with format protobuf or custom:

• The Stream Designer displays the Registered Schema ID selector populated from the registry.
• For Protobuf schemas, the Proto3 Message Type dropdown populates with extracted messages.
• The Autofill Fields 🪄 button maps Protobuf field types to rekuiper stream types automatically:
  • string -> string (text)
  • double, float -> float (64-bit double)
  • int32, int64, uint32, uint64 -> bigint (64-bit integer)
  • bool -> boolean
  • bytes -> bytea (raw byte buffer)
  • repeated <type> -> array
• The Generated DDL SQL automatically updates with SCHEMAID = "<id>" and MESSAGETYPE = "<type>".

3. Rule Designer Multi-Protocol Egress

Rule Designer Sinks support four serialization modes:

• JSON: Default serialized object or array.
• Protobuf 3: Serializes outgoing alert or telemetry events into binary protobuf payloads matching the schema contract.
• Custom Codec: Applies custom binary encoders.
• Delimited: Encodes egress rows using specified delimiter characters (e.g. comma or tab).

4. Reference Tracking and Safe Deletion Guard

The registry tracks all streams, scan tables, and rule egress actions that reference each schema.

• If a schema has active references, the registry displays an active reference badge.
• If a user attempts to drop an active schema, the registry displays a warning dialog listing all dependent pipelines before deletion.

---

REST API Reference

The Schema Registry adheres to standard eKuiper REST endpoints:

Method	Endpoint	Description
GET	/schemas/protobuf	List all registered Protobuf schema names
GET	/schemas/protobuf/:name	Get Protobuf schema definition and text content
POST	/schemas/protobuf	Register new Protobuf schema ({ name, content })
PUT	/schemas/protobuf/:name	Update existing Protobuf schema text content
DELETE	/schemas/protobuf/:name	Drop Protobuf schema from registry
GET	/schemas/custom	List all registered custom codec schemas
GET	/schemas/custom/:name	Get custom codec schema definition
POST	/schemas/custom	Register custom codec schema
PUT	/schemas/custom/:name	Update custom codec schema
DELETE	/schemas/custom/:name	Drop custom codec schema

---

Example Usage

1. Register a Protobuf Schema via REST

curl -X POST http://localhost:9081/schemas/protobuf \
  -H "Content-Type: application/json" \
  -d '{
    "name": "factory_telemetry",
    "content": "syntax = \"proto3\";\npackage factory;\n\nmessage SensorReading {\n  string device_id = 1;\n  double temperature = 2;\n  double pressure = 3;\n  int64 timestamp = 4;\n}"
  }'

2. Create a Stream Using the Schema in SQL DDL

CREATE STREAM factory_stream (
    device_id string,
    temperature float,
    pressure float,
    timestamp bigint
) WITH (
    DATASOURCE = "factory/sensors",
    FORMAT = "protobuf",
    SCHEMAID = "factory_telemetry",
    MESSAGETYPE = "SensorReading",
    TYPE = "mqtt"
);

3. Configure a Rule with Protobuf Serialization Sink

{
  "id": "rule_proto_alert",
  "sql": "SELECT device_id, temperature FROM factory_stream WHERE temperature > 85.0",
  "actions": [
    {
      "mqtt": {
        "server": "tcp://127.0.0.1:1883",
        "topic": "factory/alerts/proto",
        "format": "protobuf",
        "schemaId": "factory_telemetry",
        "messageType": "SensorReading",
        "sendSingle": true
      }
    }
  ]
}