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.

Opt-in default. The connector is disabled by default (Connectors.Gcp.Enable = false). A stock kubemq-server does not bind gRPC port 8085 until you turn it on. Enable it with its enable variable before pointing clients at port 8085:

docker run -d \  --name kubemq \  -p 8085:8085 \  -p 50000:50000 \  -e KUBEMQ_TOKEN=YOUR_LICENSE_KEY \  -e CONNECTORS_GCP_ENABLE=true \  europe-docker.pkg.dev/kubemq/images/kubemq:next

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

The GCP connector runs in emulator mode: there is no OAuth2 validation, no JWT verification, no TLS, and IAM RPCs (GetIamPolicy, SetIamPolicy, TestIamPermissions) are permissive stubs that echo requests without enforcement.

• No authentication. All connecting clients are trusted unconditionally.
• No TLS. The gRPC listener is plaintext. 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

This recipe confirms that a basic publish → consume round-trip reaches KubeMQ and that fan-out to both the Events Store log and the subscription Queue is working.

Prerequisites: KubeMQ running with the GCP connector enabled (CONNECTORS_GCP_ENABLE=true) and port 8085 reachable.

export PUBSUB_EMULATOR_HOST=localhost:8085
export PUBSUB_PROJECT_ID=smoke-test

import os, time
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"

pub = pubsub_v1.PublisherClient()
sub = pubsub_v1.SubscriberClient()

t = pub.topic_path("smoke-test", "smoke-topic")         # -> gcp.smoke-topic
s = sub.subscription_path("smoke-test", "smoke-sub")    # -> gcp.sub.smoke-sub

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

# Publish one message.
future = pub.publish(t, data=b"smoke-payload", env="ci")
msg_id = future.result()
print(f"published id={msg_id}")

# Pull and confirm arrival.
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.close()

Cross-protocol fan-out check

The headline behavior of the connector is that a Pub/Sub publish is also a native KubeMQ message. After the publish above, the same message is on the Events Store log gcp.smoke-topic, and the subscription backlog is a 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