Migrating from Google Cloud Pub/Sub

Point your existing Google Cloud Pub/Sub application at KubeMQ by changing one environment variable. The KubeMQ GCP connector exposes an emulator-compatible Pub/Sub v1 gRPC surface, so unmodified google-cloud-pubsub 2.x clients, gcloud pubsub, and any SDK that honours PUBSUB_EMULATOR_HOST connect with no application code change — set the env var, clear the Google credentials, and you are on KubeMQ. There is no KubeMQ SDK to adopt, no proto, and no data migration: topics live in normal KubeMQ Events Store logs and subscriptions in normal Queue channels. Rollback is config-only (CONNECTORS_GCP_ENABLE=false).

This is an endpoint-only drop-in. But several connector behaviors deviate from real Google Pub/Sub — read What Does NOT Migrate / Deviations before you cut over; most are invisible until a corner case hits production.

Overview

KubeMQ's GCP connector exposes an emulator-compatible gRPC surface that accepts unmodified google-cloud-pubsub 2.x client libraries, gcloud pubsub, and any other SDK that honours PUBSUB_EMULATOR_HOST. No application code changes are required — point the env var at KubeMQ and clear the Google credentials.

Quick start: repoint one client

Try the switch on your own machine before you plan the migration.

The Docker tab points one client at the KubeMQ you started with Try KubeMQ, turning its connector on where needed and keeping your evaluation or license and your data. The Kubernetes tab works on a cluster you installed with Install on Kubernetes. Use it to try the switch, not for production; see Plans compared for staging and production licenses. If your current broker runs on this machine, stop it first, because two programs cannot publish the same port. The kmq command runs natively on Windows; the other commands are bash, so on Windows run them in WSL (Windows Subsystem for Linux).

kmq deploy update --installation kubemq --connector gcp --deadline 10m

{
  "installation_id": "…",
  "state": "…",
  "target": "…",
  "data_retained": true,
  "next_action": "…"
}

export GOOGLE_APPLICATION_CREDENTIALS=/path/to/key.json
export GOOGLE_CLOUD_PROJECT=my-project

export PUBSUB_EMULATOR_HOST=localhost:8085
export PUBSUB_PROJECT_ID=my-project   # any value; the connector ignores the project
unset GOOGLE_APPLICATION_CREDENTIALS

python3 -m pip install google-cloud-pubsub

import os, time, uuid
from google.cloud import pubsub_v1          # google-cloud-pubsub 2.x

os.environ["PUBSUB_EMULATOR_HOST"] = "localhost:8085"
os.environ["PUBSUB_PROJECT_ID"]    = "smoke-test"

suffix = uuid.uuid4().hex[:8]               # new names on every run
pub = pubsub_v1.PublisherClient()
sub = pubsub_v1.SubscriberClient()

t = pub.topic_path("smoke-test", f"smoke-topic-{suffix}")       # -> gcp.smoke-topic-<suffix>
s = sub.subscription_path("smoke-test", f"smoke-sub-{suffix}")  # -> gcp.sub.smoke-sub-<suffix>

pub.create_topic(request={"name": t})
sub.create_subscription(request={"name": s, "topic": t})

pub.publish(t, data=b"smoke-payload", env="ci").result()
print("published")

time.sleep(0.5)   # allow fan-out
resp = sub.pull(request={"subscription": s, "max_messages": 1})
assert len(resp.received_messages) == 1, "expected 1 message"
rm = resp.received_messages[0]
assert rm.message.data == b"smoke-payload"
assert rm.message.attributes["env"] == "ci"
sub.acknowledge(request={"subscription": s, "ack_ids": [rm.ack_id]})
print("smoke test PASSED — message received and acked")

sub.delete_subscription(request={"subscription": s})
pub.delete_topic(request={"topic": t})
sub.close()
print("cleaned up")

python3 smoke_test.py

published
smoke test PASSED — message received and acked
cleaned up

gcp:
  enabled: true
  sessionAffinity: ClientIP

kmq deploy update --installation messaging --kube-context YOUR_KUBE_CONTEXT --values cluster-values.yaml --deadline 10m

{
  "installation_id": "…",
  "state": "…",
  "target": "…",
  "data_retained": true,
  "next_action": "…"
}

kubectl --context YOUR_KUBE_CONTEXT -n kubemq get svc messaging-gcp

NAME             TYPE        CLUSTER-IP   EXTERNAL-IP   PORT(S)     AGE
messaging-gcp    ClusterIP   …            <none>        8085/TCP    …

kubectl --context YOUR_KUBE_CONTEXT -n kubemq port-forward svc/messaging-gcp 8085:8085

Forwarding from 127.0.0.1:8085 -> 8085
Forwarding from [::1]:8085 -> 8085

Before you move production traffic

• Read What does not migrate before you commit.
• Plan how traffic moves and how you roll back: run both brokers, move one subscription at a time, and keep Google Cloud Pub/Sub until you have verified the move.
• Install KubeMQ for real: Choose your path.

Compatibility Matrix

¹ Dead-lettering is connector-level, not a broker redrive. When a message's receive count exceeds max_delivery_attempts, the connector republishes it to the configured dead-letter topic through the normal fan-out path. This assigns new message IDs and resets the delivery counter; the broker's own queue-redrive path is not involved. max_delivery_attempts must be 5..100; leaving it unset disables the dead-letter policy.

Connection / Endpoint Migration

The only required change is the endpoint env var. The SDK clears Google credentials and uses the insecure emulator path automatically.

Before (real Google Cloud)

# Application authenticates with ADC or a service-account key.
export GOOGLE_APPLICATION_CREDENTIALS=/path/to/key.json
export GOOGLE_CLOUD_PROJECT=my-project

from google.cloud import pubsub_v1

publisher = pubsub_v1.PublisherClient()
subscriber = pubsub_v1.SubscriberClient()

After (KubeMQ)

# Point every Google Pub/Sub SDK at KubeMQ.
# Clear credentials so the library enters unauthenticated emulator mode.
export PUBSUB_EMULATOR_HOST=kubemq-host:8085
export PUBSUB_PROJECT_ID=my-project   # arbitrary; the connector ignores the project segment

unset GOOGLE_APPLICATION_CREDENTIALS
unset GOOGLE_CLOUD_PROJECT

No other client-code change is needed. The connector listens on port 8085 — the same port Google's emulator uses — so an existing PUBSUB_EMULATOR_HOST that already points at the emulator only needs its host changed. The connector validates and strips the projects/{p}/ prefix from all resource paths but otherwise treats resource IDs as global (single-tenant, like the emulator).

Optional — enable AdvertisedEndpoint so the dashboard shows the correct PUBSUB_EMULATOR_HOST hint:

[Connectors.Gcp]
Enable = true
Port   = "8085"
AdvertisedEndpoint = "kubemq.mycompany.svc:8085"

Concept & Destination Mapping

Fan-out model. A Publish call writes the message exactly once to the Events Store log gcp.{t}, then fans out one Queue message per subscription (applying each subscription's filter at publish time). This means:

• A Pub/Sub topic publish is immediately visible to native KubeMQ consumers on Events Store channel gcp.{t}.
• A subscription's unconsumed backlog is a native KubeMQ Queue on channel gcp.sub.{s}.

Resource ID rules (same as Google's): 3..255 chars, must start with a letter, charset [A-Za-z0-9._~%+-], no goog prefix, topic IDs may not start with sub. (reserved broker namespace).

Message attribute pass-through

A PubsubMessage with attributes and an ordering_key arrives at native consumers with three reserved tags added:

These tags are stripped from attributes when the message is delivered back to a Pub/Sub SDK client.

Canonical Client Example

Client: google-cloud-pubsub 2.x (Python pubsub_v1 gRPC stubs).

import os
from google.cloud import pubsub_v1          # google-cloud-pubsub 2.x

# ── Environment ─────────────────────────────────────────────────────────────
os.environ["PUBSUB_EMULATOR_HOST"] = "localhost:8085"
os.environ["PUBSUB_PROJECT_ID"]    = "my-project"

PROJECT = "my-project"
TOPIC   = "orders"
SUB     = "orders-sub"

# ── Create topic and subscription ────────────────────────────────────────────
publisher  = pubsub_v1.PublisherClient()          # honours PUBSUB_EMULATOR_HOST
subscriber = pubsub_v1.SubscriberClient()

topic_path = publisher.topic_path(PROJECT, TOPIC)        # -> gcp.orders
sub_path   = subscriber.subscription_path(PROJECT, SUB)  # -> gcp.sub.orders-sub

publisher.create_topic(request={"name": topic_path})
subscriber.create_subscription(
    request={"name": sub_path, "topic": topic_path}
)

# ── Publish ───────────────────────────────────────────────────────────────────
future = publisher.publish(           # pubsub_v1.PublisherClient.publish
    topic_path,
    data=b"order-001",                # bytes
    region="eu-west-1",               # arbitrary attributes (passed as Tags)
)
print("published:", future.result())  # blocks until the broker acks

# ── Pull (synchronous) ────────────────────────────────────────────────────────
response = subscriber.pull(           # pubsub_v1.SubscriberClient.pull
    request={"subscription": sub_path, "max_messages": 10}
)
for msg in response.received_messages:
    print("received:", msg.message.data, msg.message.attributes)
    subscriber.acknowledge(           # pubsub_v1.SubscriberClient.acknowledge
        request={"subscription": sub_path, "ack_ids": [msg.ack_id]}
    )

# ── StreamingPull (async callback) ───────────────────────────────────────────
def callback(message: pubsub_v1.subscriber.message.Message) -> None:
    print("streaming:", message.data, message.attributes)
    message.ack()                     # pubsub_v1.subscriber.message.Message.ack

streaming_future = subscriber.subscribe(sub_path, callback=callback)
try:
    streaming_future.result(timeout=10)
except Exception:
    streaming_future.cancel()
    streaming_future.result()

subscriber.close()

RPC is not a Pub/Sub primitive

Pub/Sub has no request/reply mechanism. If you need RPC, use KubeMQ's native Commands/Queries pattern over gRPC or REST rather than modelling it with a reply topic.

Security

Google IAM and OAuth2 are not modelled; the IAM calls (GetIamPolicy, SetIamPolicy, TestIamPermissions) are permissive stubs. When KubeMQ authentication is on, every call must carry a KubeMQ token in the authorization metadata (a Bearer prefix is accepted). When it is off, any client that reaches the port is trusted. The listener has no TLS.

• TLS. For encrypted transport, terminate TLS at a reverse proxy or service mesh in front of port 8085.
• IAM stubs. GetIamPolicy returns an empty Policy{Version: 3}. No permissions are checked.
• DoS guards remain active. MaxRecvMsgSize, the per-subscription in-flight cap (MaxInflightPerSubscription), MaxConcurrentPolls, MaxSeekReplay, and push delivery backoff are enforced regardless of auth mode.

What Does NOT Migrate / Deviations

Features not supported (hard rejections)

Behavioral deviations

Verification Smoke Test

For the basic round trip, see the quick start; this section adds deeper checks.

Cross-protocol fan-out check

The headline behavior of the connector is that a Pub/Sub publish is also a native KubeMQ message. Create a topic smoke-topic and a subscription smoke-sub with the same client calls the quick start uses, but with fixed names and no cleanup, so a native KubeMQ client can read them back:

t = pub.topic_path("smoke-test", "smoke-topic")
s = sub.subscription_path("smoke-test", "smoke-sub")
pub.create_topic(request={"name": t})
sub.create_subscription(request={"name": s, "topic": t})

A publish to this topic lands on the Events Store log gcp.smoke-topic, and the backlog of this subscription is the native Queue gcp.sub.smoke-sub. A native KubeMQ client confirms the fan-out with no Pub/Sub SDK involved:

# Events Store log carries the topic publish:
#
#   SubscribeToEventsStore(channel="gcp.smoke-topic", startAt="new")
#
# Subscription backlog is a native Queue channel:
#
#   ReceiveQueueMessages(channel="gcp.sub.smoke-sub", maxMessages=10)

For a deterministic read, subscribe to the Events Store log with start policy startAt = "new" before publishing. See Channel Mapping for the full gcp.{topic} / gcp.sub.{subscription} scheme.

See Also