KubeMQ
ConnectorsGoogle Cloud Pub/SubHow-to guides

Schema Validation

Avro and Protobuf schema enforcement over KubeMQ — CreateSchema, topic schema_settings, enforce-on-publish, ≤300 KB definitions, and revisions.

A schema describes the shape every message published to a topic must conform to. When a topic references a schema, the connector enforces it on publish — non-conforming messages are rejected before they ever reach the topic log. The connector supports the two schema types Google Cloud Pub/Sub supports: Avro and Protobuf, stored as records in a per-node replicated registry with a full revision history. The SchemaService ships 10 RPCs.

How enforcement works

  1. CreateSchema registers a schema definition — an Avro JSON/IDL definition or a Protobuf message definition — under a schema id. The definition must be ≤ 300 KB.
  2. A topic is created (or updated) with schema_settings referencing that schema, plus an encoding — JSON or BINARY.
  3. On every Publish to that topic the connector validates each message's data against the schema. Validation is part of the atomic batch check: the whole batch is rejected (INVALID_ARGUMENT) on the first non-conforming message — nothing in the batch is enqueued.
CreateSchema(avro|protobuf, ≤ 300 KB) ──▶ registry record (revisions)
CreateTopic(schema_settings → schema id, encoding) ──▶ topic bound to schema
Publish(batch) ──▶ validate each message against the schema
        first non-conforming message ──▶ reject WHOLE batch (INVALID_ARGUMENT)
        all conforming                ──▶ write once to gcp.{t}, then fan out

A conforming publish then follows the normal path — written once to the Events Store log gcp.{t} and fanned out one queue copy per subscription. See Publishing.

Enforcement is all-or-nothing per batch. Because publish is atomic, one bad message rejects the entire Publish call and enqueues nothing. Validate client-side, or publish smaller batches, if you want finer-grained failure isolation.

Defining and binding a schema

Register the definition, then bind a topic to it with schema_settings:

schema, _ := schemaClient.CreateSchema(ctx, &pubsubpb.CreateSchemaRequest{
	Parent:   "projects/" + projectID,
	SchemaId: "order-v1",
	Schema:   &pubsubpb.Schema{Type: pubsubpb.Schema_AVRO, Definition: avroDef},
})
_, _ = client.CreateTopic(ctx, &pubsubpb.Topic{
	Name:           "projects/" + projectID + "/topics/orders",
	SchemaSettings: &pubsubpb.SchemaSettings{Schema: schema.Name, Encoding: pubsubpb.Encoding_JSON},
})
from google.cloud import pubsub_v1
from google.pubsub_v1.types import Schema, Encoding

schema = schema_client.create_schema(
    request={"parent": f"projects/{project_id}", "schema_id": "order-v1",
             "schema": {"type_": Schema.Type.AVRO, "definition": avro_def}})
publisher.create_topic(request={"name": topic_path,
    "schema_settings": {"schema": schema.name, "encoding": Encoding.JSON}})
Schema schema = schemaClient.createSchema(
    SchemaName.of(projectId, "order-v1").getParent(),
    Schema.newBuilder().setType(Schema.Type.AVRO).setDefinition(avroDef).build(),
    "order-v1");
topicAdminClient.createTopic(Topic.newBuilder()
    .setName(topicName.toString())
    .setSchemaSettings(SchemaSettings.newBuilder()
        .setSchema(schema.getName()).setEncoding(Encoding.JSON).build())
    .build());
const [schema] = await pubsub.createSchema('order-v1', SchemaTypes.Avro, avroDef);
await pubsub.createTopic({
  name: 'orders',
  schemaSettings: { schema: schema.name, encoding: 'JSON' },
});
var schema = await schemaClient.CreateSchemaAsync(
    new ProjectName(projectId), new Schema { Type = Schema.Types.Type.Avro, Definition = avroDef },
    "order-v1");
await publisher.CreateTopicAsync(new Topic
{
    TopicName = TopicName.FromProjectTopic(projectId, "orders"),
    SchemaSettings = new SchemaSettings { Schema = schema.Name, Encoding = Encoding.Json },
});
schema = schema_client.create_schema parent: "projects/#{project_id}",
  schema_id: "order-v1", schema: { type: :AVRO, definition: avro_def }
publisher.create_topic name: topic_path,
  schema_settings: { schema: schema.name, encoding: :JSON }

Avro vs Protobuf

TypeHow it is validatedNotes
Avroparsed and validated via the connector's Avro enginethe definition is an Avro JSON / IDL definition
Protobufparsed and validated via the connector's protoreflect enginethe definition is a proto message definition

Both enforce at publish time with identical batch-atomic semantics; the only difference is the definition language and the encoding. A definition that fails to parse — for either type — is rejected with INVALID_ARGUMENT at CreateSchema / CommitSchema time.

Revisions

A schema is versioned, and the connector keeps a full revision history:

  • CommitSchema adds a new revision to an existing schema.
  • RollbackSchema creates a new revision that restores a prior definition.
  • DeleteSchemaRevision removes a revision but always keeps at least one — you cannot delete the last remaining revision.
  • ListSchemaRevisions / GetSchema (BASIC or FULL) read the history; ValidateSchema and ValidateMessage check a definition or a payload without publishing.

A producer can evolve its schema across revisions without breaking topics that have not yet opted in to the new revision.

Schema definitions are capped at 300 KB. A CreateSchema or CommitSchema with a definition larger than 300 KB is rejected. Keep schemas focused. See Limits & rules.

Publishing

Publish atomicity — the batch validation that schema enforcement plugs into.

Capabilities

The 10 SchemaService RPCs and the full supported v1 surface.

Limits & rules

The ≤ 300 KB schema cap and the publish batch validation rules.

Was this page helpful?

Reliability

Delivery guarantees over KubeMQ — dead-letter topics, retry backoff, at-least-once redelivery, node-local exactly-once, and retention clamped to the broker.

Seek & Snapshots

Replay over KubeMQ — Seek to a timestamp or snapshot, replay bounded by MaxSeekReplay, pre-window timestamps clamped to earliest, and snapshot expiry.

On this page

How enforcement worksDefining and binding a schemaAvro vs ProtobufRevisionsRelated