# Schema Validation (/connectors/gcp-pub-sub/how-to/schema-validation)



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 [#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.

```text
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](/connectors/gcp-pub-sub/how-to/publishing).

<Callout type="warn">
  **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.
</Callout>

## Defining and binding a schema [#defining-and-binding-a-schema]

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

<Tabs groupId="language" items="['Go','Python','Java','JavaScript','C#','Ruby']">
  <Tab value="Go">
    ```go
    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},
    })
    ```
  </Tab>

  <Tab value="Python">
    ```python
    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}})
    ```
  </Tab>

  <Tab value="Java">
    ```java
    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());
    ```
  </Tab>

  <Tab value="JavaScript">
    ```javascript
    const [schema] = await pubsub.createSchema('order-v1', SchemaTypes.Avro, avroDef);
    await pubsub.createTopic({
      name: 'orders',
      schemaSettings: { schema: schema.name, encoding: 'JSON' },
    });
    ```
  </Tab>

  <Tab value="C#">
    ```csharp
    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 },
    });
    ```
  </Tab>

  <Tab value="Ruby">
    ```ruby
    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 }
    ```
  </Tab>
</Tabs>

## Avro vs Protobuf [#avro-vs-protobuf]

| Type         | How it is validated                                          | Notes                                           |
| ------------ | ------------------------------------------------------------ | ----------------------------------------------- |
| **Avro**     | parsed and validated via the connector's Avro engine         | the definition is an Avro JSON / IDL definition |
| **Protobuf** | parsed and validated via the connector's protoreflect engine | the 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 [#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.

<Callout type="warn">
  **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](/connectors/gcp-pub-sub/reference/limits-and-rules).
</Callout>

## Related [#related]

<Cards>
  <Card title="Publishing" href="/connectors/gcp-pub-sub/how-to/publishing" description="Publish atomicity — the batch validation that schema enforcement plugs into." />

  <Card title="Capabilities" href="/connectors/gcp-pub-sub/reference/capabilities" description="The 10 SchemaService RPCs and the full supported v1 surface." />

  <Card title="Limits & rules" href="/connectors/gcp-pub-sub/reference/limits-and-rules" description="The ≤ 300 KB schema cap and the publish batch validation rules." />
</Cards>