# Ordered Delivery (/connectors/gcp-pub-sub/how-to/ordered-delivery)



**Ordered delivery** guarantees that messages sharing an **ordering key** are delivered in publish order, with **at most one in flight per key**. The head of a key blocks until it is acknowledged (or redelivered), and redelivery is in order — so a per-customer or per-aggregate stream is processed strictly in sequence. Messages without an ordering key are delivered unordered, and independent keys make progress in parallel. Ordering is a per-message feature with no code path beyond the standard Pub/Sub ordering-key API; the connector enforces the per-key sequencing internally.

## Overview [#overview]

Ordering is enabled on **both** sides: set `enable_message_ordering` on the publisher (it serializes publishes per key) and on the subscription (it enforces one-in-flight-per-key delivery). Each publish then carries an `ordering_key`; the key rides across the wire as the reserved tag `_pubsub_ordering_key`, which the connector surfaces as the message's ordering key for Pub/Sub clients.

| Step            | Action                                                 | Behavior                                                   |
| --------------- | ------------------------------------------------------ | ---------------------------------------------------------- |
| Topic           | `CreateTopic` + publisher ordering enabled             | Publisher serializes publishes per key                     |
| Subscription    | `CreateSubscription(enable_message_ordering=true)`     | One copy in flight per key                                 |
| Publish         | `Publish(ordering_key="cust-7")`                       | In-order within the key; carried as `_pubsub_ordering_key` |
| Pull + Ack      | `Pull` → `Acknowledge` → next per-key message released | Head-of-key blocks until acked                             |
| Keyless publish | `Publish` (no key)                                     | Unordered; not serialized                                  |

Ordering semantics:

* **Per-key total order** — within one `ordering_key`, delivery follows publish order; the next message is released only after the current one is acked or redelivered.
* **At most one in flight per key** — a key is never delivered ahead of its own un-acked head.
* **Independent keys run in parallel** — a round-robin cursor spreads delivery fairly across contended keys; a slow key never blocks a different one.
* **Redelivery stays in order** — an ack-deadline expiry or nack redelivers the head before any later message in the same key.

## How it works [#how-it-works]

The publisher serializes publishes per ordering key. The connector writes each message to the topic log `gcp.{topic}` and fans out a copy to the subscription queue `gcp.sub.{subscription}`, then releases at most one message per key at a time — the next per-key message is held until the current one is acknowledged.

<Mermaid
  chart="`
graph LR
PUB[&#x22;Pub/Sub SDK publisher<br/>(ordering enabled)&#x22;]
CONN[&#x22;Pub/Sub connector<br/>:8085&#x22;]
LOG[(&#x22;Events Store log<br/>gcp.ordered&#x22;)]
Q{{&#x22;Queue channel<br/>gcp.sub.sub-ordered&#x22;}}
BROKER[&#x22;Message Broker&#x22;]
SUB[&#x22;Pub/Sub SDK subscriber<br/>(one in flight per key)&#x22;]

PUB -- &#x22;Publish(ordering_key)&#x22; --> CONN
CONN -- &#x22;SendEventsStore&#x22; --> LOG
LOG -- &#x22;fan-out copy&#x22; --> Q
LOG --> BROKER
Q --> BROKER
BROKER -- &#x22;release head-of-key (leased)&#x22; --> CONN
CONN -- &#x22;Pull&#x22; --> SUB
SUB -. &#x22;Acknowledge → release next per-key&#x22; .-> CONN

class PUB,SUB client
class CONN connector
class LOG,Q,BROKER broker
`"
/>

*Messages sharing an `ordering_key` are delivered in publish order with at most one in flight per key; the next per-key message is released only after the current one is acknowledged, while independent keys make progress in parallel.*

## Publish and consume in order [#publish-and-consume-in-order]

Enable ordering on both the publisher and the subscription, publish interleaved messages across two keys (plus one keyless), and pull one at a time — acknowledging each before the next pull — so the per-key order is directly observable. Each client sets only `PUBSUB_EMULATOR_HOST` (default `localhost:8085`) and a project id.

<Tabs groupId="language" items="['Go','Python','Java','JavaScript','C#','Ruby']">
  <Tab value="Go">
    ```go
    package main

    import (
    	"context"
    	"fmt"
    	"log"
    	"os"
    	"time"

    	"cloud.google.com/go/pubsub"
    	"cloud.google.com/go/pubsub/apiv1/pubsubpb"
    )

    func projectID() string {
    	if v := os.Getenv("PUBSUB_PROJECT_ID"); v != "" {
    		return v
    	}
    	return "my-project"
    }

    func main() {
    	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
    	defer cancel()

    	client, err := pubsub.NewClient(ctx, projectID())
    	if err != nil {
    		log.Fatalf("NewClient: %v", err)
    	}
    	defer client.Close()

    	// CreateTopic, then enable ordering on the publisher handle (it serializes
    	// publishes per key). The subscription must enable ordering too — both sides.
    	topic, err := client.CreateTopic(ctx, "ordered")
    	if err != nil {
    		log.Fatalf("CreateTopic: %v", err)
    	}
    	topic.EnableMessageOrdering = true
    	defer topic.Stop()

    	if _, err := client.CreateSubscription(ctx, "sub-ordered", pubsub.SubscriptionConfig{
    		Topic:                 topic,
    		AckDeadline:           10 * time.Second,
    		EnableMessageOrdering: true,
    	}); err != nil {
    		log.Fatalf("CreateSubscription: %v", err)
    	}

    	// Two keys interleaved + one keyless. Await each result so the per-key publish
    	// order is preserved across keys.
    	plan := []struct{ key, body string }{
    		{"cust-7", "A1"}, {"cust-9", "B1"}, {"cust-7", "A2"},
    		{"", "keyless"}, {"cust-9", "B2"}, {"cust-7", "A3"}, {"cust-9", "B3"},
    	}
    	for _, p := range plan {
    		m := &pubsub.Message{Data: []byte(p.body)}
    		if p.key != "" {
    			m.OrderingKey = p.key
    		}
    		if _, err := topic.Publish(ctx, m).Get(ctx); err != nil {
    			log.Fatalf("Publish %q: %v", p.body, err)
    		}
    	}

    	// Pull one at a time, ack before the next pull → connector releases the next
    	// per-key message in order.
    	subClient, err := pubsub.NewSubscriberClient(ctx)
    	if err != nil {
    		log.Fatalf("NewSubscriberClient: %v", err)
    	}
    	defer subClient.Close()
    	subPath := fmt.Sprintf("projects/%s/subscriptions/%s", projectID(), "sub-ordered")

    	for i := 0; i < len(plan); i++ {
    		resp, err := subClient.Pull(ctx, &pubsubpb.PullRequest{Subscription: subPath, MaxMessages: 1})
    		if err != nil {
    			log.Fatalf("Pull: %v", err)
    		}
    		if len(resp.GetReceivedMessages()) == 0 {
    			i--
    			continue
    		}
    		rm := resp.GetReceivedMessages()[0]
    		key := rm.GetMessage().GetOrderingKey() // surfaced from _pubsub_ordering_key.
    		fmt.Printf("received body=%q ordering_key=%q\n", string(rm.GetMessage().GetData()), key)
    		_ = subClient.Acknowledge(ctx, &pubsubpb.AcknowledgeRequest{
    			Subscription: subPath, AckIds: []string{rm.GetAckId()},
    		})
    	}
    	// Per-key order is preserved: cust-7 → A1,A2,A3 and cust-9 → B1,B2,B3.
    }
    ```
  </Tab>

  <Tab value="Python">
    ```python
    import os

    from google.cloud import pubsub_v1
    from google.cloud.pubsub_v1.types import PublisherOptions


    def project_id() -> str:
        return os.environ.get("PUBSUB_PROJECT_ID", "my-project")


    def main() -> None:
        proj = project_id()
        # The publisher MUST enable message ordering (it serializes publishes per key).
        publisher = pubsub_v1.PublisherClient(
            publisher_options=PublisherOptions(enable_message_ordering=True)
        )
        subscriber = pubsub_v1.SubscriberClient()
        topic_path = publisher.topic_path(proj, "ordered")          # -> gcp.ordered
        sub_path = subscriber.subscription_path(proj, "sub-ordered")  # -> gcp.sub.sub-ordered

        publisher.create_topic(request={"name": topic_path})
        # The subscription must also enable ordering.
        subscriber.create_subscription(
            request={"name": sub_path, "topic": topic_path, "enable_message_ordering": True}
        )

        # Two keys interleaved + one keyless; await each publish to preserve order.
        plan = [("cust-7", "A1"), ("cust-9", "B1"), ("cust-7", "A2"),
                ("", "keyless"), ("cust-9", "B2"), ("cust-7", "A3"), ("cust-9", "B3")]
        for key, body in plan:
            publisher.publish(topic_path, body.encode(), ordering_key=key).result(timeout=15)

        # Pull one at a time, ack before the next pull → next per-key message released.
        for _ in range(len(plan)):
            resp = subscriber.pull(request={"subscription": sub_path, "max_messages": 1}, timeout=20)
            if not resp.received_messages:
                continue
            rm = resp.received_messages[0]
            print(f"received body={rm.message.data.decode()!r} ordering_key={rm.message.ordering_key!r}")
            subscriber.acknowledge(request={"subscription": sub_path, "ack_ids": [rm.ack_id]})
        subscriber.close()
        # cust-7 → A1,A2,A3 and cust-9 → B1,B2,B3 in order.


    if __name__ == "__main__":
        main()
    ```
  </Tab>

  <Tab value="Java">
    ```java
    import com.google.api.gax.core.NoCredentialsProvider;
    import com.google.api.gax.grpc.GrpcTransportChannel;
    import com.google.api.gax.rpc.FixedTransportChannelProvider;
    import com.google.api.gax.rpc.TransportChannelProvider;
    import com.google.cloud.pubsub.v1.SubscriptionAdminClient;
    import com.google.cloud.pubsub.v1.SubscriptionAdminSettings;
    import com.google.cloud.pubsub.v1.TopicAdminClient;
    import com.google.cloud.pubsub.v1.TopicAdminSettings;
    import com.google.cloud.pubsub.v1.stub.GrpcSubscriberStub;
    import com.google.cloud.pubsub.v1.stub.SubscriberStubSettings;
    import com.google.protobuf.ByteString;
    import com.google.pubsub.v1.AcknowledgeRequest;
    import com.google.pubsub.v1.PublishRequest;
    import com.google.pubsub.v1.PubsubMessage;
    import com.google.pubsub.v1.PullRequest;
    import com.google.pubsub.v1.PullResponse;
    import com.google.pubsub.v1.PushConfig;
    import com.google.pubsub.v1.ReceivedMessage;
    import com.google.pubsub.v1.Subscription;
    import com.google.pubsub.v1.SubscriptionName;
    import com.google.pubsub.v1.TopicName;
    import io.grpc.ManagedChannel;
    import io.grpc.ManagedChannelBuilder;
    import java.util.List;

    public final class Main {
        public static void main(String[] args) throws Exception {
            String emulatorHost = System.getenv().getOrDefault("PUBSUB_EMULATOR_HOST", "localhost:8085");
            String projectId = System.getenv().getOrDefault("PUBSUB_PROJECT_ID", "my-project");

            ManagedChannel channel = ManagedChannelBuilder.forTarget(emulatorHost).usePlaintext().build();
            TransportChannelProvider channelProvider =
                    FixedTransportChannelProvider.create(GrpcTransportChannel.create(channel));
            NoCredentialsProvider noCreds = NoCredentialsProvider.create();

            TopicName topic = TopicName.of(projectId, "ordered");                 // -> gcp.ordered
            SubscriptionName sub = SubscriptionName.of(projectId, "sub-ordered"); // -> gcp.sub.sub-ordered

            try (TopicAdminClient topicAdmin = TopicAdminClient.create(TopicAdminSettings.newBuilder()
                    .setTransportChannelProvider(channelProvider).setCredentialsProvider(noCreds).build());
                    SubscriptionAdminClient subAdmin = SubscriptionAdminClient.create(
                            SubscriptionAdminSettings.newBuilder()
                                    .setTransportChannelProvider(channelProvider)
                                    .setCredentialsProvider(noCreds).build());
                    GrpcSubscriberStub subStub = GrpcSubscriberStub.create(SubscriberStubSettings.newBuilder()
                            .setTransportChannelProvider(channelProvider)
                            .setCredentialsProvider(noCreds).build())) {

                topicAdmin.createTopic(topic);
                // The subscription enables ordering; the publisher orders per key below.
                subAdmin.createSubscription(Subscription.newBuilder()
                        .setName(sub.toString()).setTopic(topic.toString())
                        .setAckDeadlineSeconds(10).setEnableMessageOrdering(true).build());

                // Two keys interleaved + one keyless; publish in order with an ordering key.
                String[][] plan = {
                    {"cust-7", "A1"}, {"cust-9", "B1"}, {"cust-7", "A2"},
                    {"", "keyless"}, {"cust-9", "B2"}, {"cust-7", "A3"}, {"cust-9", "B3"},
                };
                for (String[] p : plan) {
                    PubsubMessage.Builder m = PubsubMessage.newBuilder()
                            .setData(ByteString.copyFromUtf8(p[1]));
                    if (!p[0].isEmpty()) {
                        m.setOrderingKey(p[0]);
                    }
                    topicAdmin.publish(PublishRequest.newBuilder()
                            .setTopic(topic.toString()).addMessages(m.build()).build());
                }

                // Pull one at a time, ack before the next pull → next per-key message released.
                for (int i = 0; i < plan.length; i++) {
                    PullResponse resp = subStub.pullCallable().call(PullRequest.newBuilder()
                            .setSubscription(sub.toString()).setMaxMessages(1).build());
                    if (resp.getReceivedMessagesCount() == 0) {
                        i--;
                        continue;
                    }
                    ReceivedMessage got = resp.getReceivedMessages(0);
                    System.out.printf("received body=%s ordering_key=%s%n",
                            got.getMessage().getData().toStringUtf8(), got.getMessage().getOrderingKey());
                    subStub.acknowledgeCallable().call(AcknowledgeRequest.newBuilder()
                            .setSubscription(sub.toString()).addAckIds(got.getAckId()).build());
                }
                // cust-7 → A1,A2,A3 and cust-9 → B1,B2,B3 in order.
                List.of(); // no-op to keep imports tidy
            } finally {
                channel.shutdown();
            }
        }
    }
    ```
  </Tab>

  <Tab value="JavaScript">
    ```typescript
    import { PubSub, v1 } from "@google-cloud/pubsub";

    const projectId = process.env["PUBSUB_PROJECT_ID"] ?? "my-project";
    const baseOptions = new PubSub({ projectId }).options;
    const options = { ...baseOptions, port: baseOptions.port === undefined ? undefined : Number(baseOptions.port) };
    const publisher = new v1.PublisherClient(options);
    const subscriber = new v1.SubscriberClient(options);

    async function main(): Promise<void> {
      const topic = publisher.projectTopicsPath(projectId, "ordered");   // -> gcp.ordered
      const sub = subscriber.subscriptionPath(projectId, "sub-ordered"); // -> gcp.sub.sub-ordered

      await publisher.createTopic({ name: topic });
      // The subscription enables ordering; each publish below carries an orderingKey.
      await subscriber.createSubscription({ name: sub, topic, ackDeadlineSeconds: 10, enableMessageOrdering: true });

      // Two keys interleaved + one keyless; publish each in turn to preserve order.
      const plan: Array<[string, string]> = [
        ["cust-7", "A1"], ["cust-9", "B1"], ["cust-7", "A2"],
        ["", "keyless"], ["cust-9", "B2"], ["cust-7", "A3"], ["cust-9", "B3"],
      ];
      for (const [orderingKey, body] of plan) {
        await publisher.publish({ topic, messages: [{ data: Buffer.from(body), orderingKey }] });
      }

      // Pull one at a time, ack before the next pull → next per-key message released.
      for (let i = 0; i < plan.length; i++) {
        const [pull] = await subscriber.pull({ subscription: sub, maxMessages: 1 });
        if (!pull.receivedMessages || pull.receivedMessages.length === 0) {
          i--;
          continue;
        }
        const rm = pull.receivedMessages[0];
        const body = Buffer.from(rm.message!.data as Uint8Array).toString("utf8");
        console.log(`received body=${body} ordering_key=${rm.message!.orderingKey}`);
        await subscriber.acknowledge({ subscription: sub, ackIds: [rm.ackId!] });
      }
      // cust-7 → A1,A2,A3 and cust-9 → B1,B2,B3 in order.
    }

    main().catch((err) => {
      console.error(err);
      process.exit(1);
    });
    ```
  </Tab>

  <Tab value="C#">
    ```csharp
    using Google.Api.Gax;
    using Google.Cloud.PubSub.V1;
    using Google.Protobuf;

    var projectId = Environment.GetEnvironmentVariable("PUBSUB_PROJECT_ID") ?? "my-project";
    var topicName = TopicName.FromProjectTopic(projectId, "ordered");                  // -> gcp.ordered
    var subName = SubscriptionName.FromProjectSubscription(projectId, "sub-ordered");  // -> gcp.sub.sub-ordered

    var publisher = await new PublisherServiceApiClientBuilder
    {
        EmulatorDetection = EmulatorDetection.EmulatorOnly,
    }.BuildAsync();
    var subscriber = await new SubscriberServiceApiClientBuilder
    {
        EmulatorDetection = EmulatorDetection.EmulatorOnly,
    }.BuildAsync();

    await publisher.CreateTopicAsync(topicName);
    // The subscription enables ordering; each publish below carries an OrderingKey.
    await subscriber.CreateSubscriptionAsync(new Subscription
    {
        SubscriptionName = subName,
        TopicAsTopicName = topicName,
        AckDeadlineSeconds = 10,
        EnableMessageOrdering = true,
    });

    // Two keys interleaved + one keyless; publish each in turn to preserve order.
    var plan = new[]
    {
        ("cust-7", "A1"), ("cust-9", "B1"), ("cust-7", "A2"),
        ("", "keyless"), ("cust-9", "B2"), ("cust-7", "A3"), ("cust-9", "B3"),
    };
    foreach (var (orderingKey, body) in plan)
    {
        var msg = new PubsubMessage { Data = ByteString.CopyFromUtf8(body) };
        if (orderingKey.Length > 0) msg.OrderingKey = orderingKey;
        await publisher.PublishAsync(topicName, new[] { msg });
    }

    // Pull one at a time, ack before the next pull → next per-key message released.
    for (var i = 0; i < plan.Length; i++)
    {
        var pull = await subscriber.PullAsync(subName, maxMessages: 1);
        if (pull.ReceivedMessages.Count == 0) { i--; continue; }
        var rm = pull.ReceivedMessages[0];
        Console.WriteLine($"received body={rm.Message.Data.ToStringUtf8()} ordering_key={rm.Message.OrderingKey}");
        await subscriber.AcknowledgeAsync(subName, new[] { rm.AckId });
    }
    // cust-7 → A1,A2,A3 and cust-9 → B1,B2,B3 in order.
    ```
  </Tab>

  <Tab value="Ruby">
    ```ruby
    # frozen_string_literal: true

    require "google/cloud/pubsub"

    project_id = ENV["PUBSUB_PROJECT_ID"] || "my-project"
    emulator_host = ENV["PUBSUB_EMULATOR_HOST"] || "localhost:8085"

    pubsub = Google::Cloud::PubSub.new(project_id: project_id, emulator_host: emulator_host)
    topic_admin = pubsub.topic_admin
    sub_admin   = pubsub.subscription_admin
    topic_path  = pubsub.topic_path("ordered")          # -> gcp.ordered
    sub_path    = pubsub.subscription_path("sub-ordered") # -> gcp.sub.sub-ordered

    topic = topic_admin.create_topic(name: topic_path)
    # The subscription enables ordering; each publish below carries an ordering_key.
    sub_admin.create_subscription(name: sub_path, topic: topic_path,
                                  ack_deadline_seconds: 10, enable_message_ordering: true)

    # Two keys interleaved + one keyless. message_ordering: true serializes per key.
    publisher = pubsub.publisher(topic.name, async: { ordered: true })
    plan = [["cust-7", "A1"], ["cust-9", "B1"], ["cust-7", "A2"],
            ["", "keyless"], ["cust-9", "B2"], ["cust-7", "A3"], ["cust-9", "B3"]]
    plan.each do |key, body|
      publisher.publish(body, ordering_key: key)
    end
    publisher.async_publisher.stop! # flush ordered publishes in order.

    # Pull one at a time, ack before the next pull → next per-key message released.
    subscriber = pubsub.subscriber(sub_path)
    plan.length.times do
      received = subscriber.pull(immediate: false, max: 1)
      next if received.empty?

      rcv = received.first
      puts "received body=#{rcv.data} ordering_key=#{rcv.ordering_key.inspect}"
      rcv.acknowledge!
    end
    # cust-7 → A1,A2,A3 and cust-9 → B1,B2,B3 in order.
    ```
  </Tab>
</Tabs>

<Callout type="info">
  **Enable ordering on both sides.** The publisher serializes publishes per key only when ordering is enabled on the publisher handle (`EnableMessageOrdering` / `enable_message_ordering` / `ordered: true`), and the subscription enforces one-in-flight-per-key only when it is created with ordering enabled. Enabling it on a single side is not enough.
</Callout>

## Why pull one at a time [#why-pull-one-at-a-time]

The examples use unary `Pull` and acknowledge each message before the next pull so the **head-of-key blocks until acked** guarantee is directly observable. A high-level streaming subscriber buffers and acks asynchronously, which preserves per-key order but obscures the strict one-in-flight-per-key sequencing. The connector enforces the order regardless; pulling one at a time only makes it visible.

## Related [#related]

<Cards>
  <Card title="Publish & Subscribe" href="/connectors/gcp-pub-sub/how-to/publish-subscribe" description="The core topic-to-subscription round-trip that ordering builds on." />

  <Card title="Subscribing" href="/connectors/gcp-pub-sub/how-to/subscribing" description="Pull vs StreamingPull, ack-deadline leases, and redelivery behavior." />

  <Card title="Channel mapping" href="/connectors/gcp-pub-sub/reference/channel-mapping" description="Per-key channels and the reserved _pubsub_ordering_key tag." />
</Cards>