Publishing

This guide covers the publish surface end to end: topic lifecycle, a single publish, batch publish (≤ 1000 messages), the atomic batch-validation rule, ordering keys, and message attributes. Every topic is a native KubeMQ Events Store log gcp.{topic} (see Channel mapping).

Topic lifecycle

The Publisher surface ships 9 RPCs (see Capabilities):

• CreateTopic — validates the name; kms_key_name is accepted-and-ignored; ingestion configs are rejected (INVALID_ARGUMENT); requested retention is clamped to the broker ceiling.
• GetTopic — returns the requested (un-clamped) retention.
• ListTopics — opaque page token.
• UpdateTopic — a FieldMask over labels, message_retention_duration, schema_settings.
• DeleteTopic — a tombstone: the record is retained so existing subscriptions survive, and re-creating the topic reuses the same log.
• ListTopicSubscriptions, ListTopicSnapshots, DetachSubscription, and Publish (below).

A single publish

Publish returns a server-assigned message id. The connector writes the message once to the Events Store log gcp.{topic} — the authoritative, cross-protocol, replayable copy and the source for Seek — then fans out one queue copy per subscription, applying each subscription's filter:

1. The SDK sends a PubsubMessage { data, attributes, ordering_key }.
2. The connector assigns a message id and a publish time and returns the id.
3. The message lands in gcp.{topic} via the Events Store send, then is fanned out to each subscription's queue gcp.sub.{s}.

A filtered-out message is never enqueued for that subscription (it is effectively auto-acked); detached subscriptions are skipped.

Batch publish

Publish accepts a batch of 1..1000 messages. Server-assigned ids are returned in request order, so a client can correlate each id with its input message.

Atomic batch validation

Per-message validation rules:

If the topic has a schema (see Schema validation), every message is also validated against it and the whole batch is rejected on the first non-conforming message. The full limit table is in Limits & rules.

A batch publish that prints its server-assigned ids in order:

from google.cloud import pubsub_v1

publisher = pubsub_v1.PublisherClient()      # honours PUBSUB_EMULATOR_HOST
topic_path = publisher.topic_path("my-project", "events")

futures = [publisher.publish(topic_path, f"event-{i}".encode()) for i in range(5)]
for fut in futures:                          # ids returned in request order
    print(fut.result())

Ordering keys

Set a per-message ordering_key and enable ordering on the subscription (enable_message_ordering). Messages sharing an ordering_key are then delivered in publish order, with at most one in flight per key — the head of a key blocks until it is acked or redelivered, and redelivery is in order. A round-robin cursor spreads delivery fairly across contended keys; keyless messages are delivered unordered.

Message attributes

A PubsubMessage's attributes map (string → string) round-trips as KubeMQ message tags. On top of the user attributes the connector carries three reserved tags across the wire:

• _pubsub_message_id — the server-assigned id;
• _pubsub_publish_time — the publish timestamp;
• _pubsub_ordering_key — the ordering key (if any).

Attribute constraints (enforced in the atomic validation above): ≤ 100 attributes; key ≤ 256 B with no goog prefix; value ≤ 1024 B.

Error quick reference

Related