Troubleshooting Pub/Sub: Common Issues and Solutions

Introduction to Pub/Sub Troubleshooting

Publish/Subscribe (Pub/Sub) messaging is a foundational pattern in modern distributed systems, enabling asynchronous communication between decoupled services. Whether you're using Google Cloud Pub/Sub, Apache Kafka, AWS SNS/SQS, or a self-hosted broker like RabbitMQ, the core architecture remains the same: publishers emit messages to topics, and subscribers receive those messages through subscriptions. While this pattern offers scalability and loose coupling, it also introduces a unique class of failure modes that can be notoriously difficult to diagnose.

This tutorial walks through the most common issues developers encounter when working with Pub/Sub systems, provides diagnostic strategies, and offers practical code examples for resolving each problem. By the end, you'll have a systematic approach to identifying and fixing Pub/Sub issues before they impact production.

Why Troubleshooting Pub/Sub Matters

Pub/Sub systems often sit at the critical path of application workflows. A delayed, dropped, or duplicated message can cascade into data inconsistency, missed business events, or degraded user experiences. Unlike synchronous HTTP calls where failures surface immediately, Pub/Sub failures can be silent — messages may appear sent but never delivered, or delivered multiple times without the publisher's knowledge.

Common consequences of unresolved Pub/Sub issues include:

• Data loss — messages published but never consumed, leading to missing records in downstream systems.
• Data duplication — the same event processed multiple times, causing double charges or corrupted state.
• Increased latency — backlog accumulation in subscriptions delaying time-sensitive operations.
• Cost overruns — retry storms and message accumulation driving up infrastructure costs.
• Cascading failures — downstream services overwhelmed by sudden message bursts after a backlog clears.

Understanding how to diagnose these issues quickly is essential for maintaining reliable event-driven architectures.

Common Issue 1: Messages Not Being Delivered

One of the most frequent and alarming issues is publishing messages that never arrive at subscribers. This can stem from several root causes, including misconfigured subscriptions, permission issues, or filter mismatches.

Diagnosing Delivery Failures

Start by verifying the message was successfully published. In Google Cloud Pub/Sub, you can check the topic's publish_message_count metric in Cloud Monitoring. If the count is zero, the problem is on the publisher side. If messages are published but not received, check the subscription's pull_request_count and ack_message_count.

Here's a diagnostic script that helps verify publishing and subscription configuration:

from google.cloud import pubsub_v1
from google.api_core.exceptions import NotFound

project_id = "your-project-id"
topic_id = "your-topic-id"
subscription_id = "your-subscription-id"

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

topic_path = publisher.topic_path(project_id, topic_id)
subscription_path = subscriber.subscription_path(project_id, subscription_id)

# Verify topic exists
try:
    topic = publisher.get_topic(request={"topic": topic_path})
    print(f"Topic exists: {topic.name}")
except NotFound:
    print("ERROR: Topic does not exist!")

# Verify subscription exists and check its config
try:
    subscription = subscriber.get_subscription(
        request={"subscription": subscription_path}
    )
    print(f"Subscription exists: {subscription.name}")
    print(f"Topic for subscription: {subscription.topic}")
    print(f"Ack deadline: {subscription.ack_deadline_seconds}s")
    if subscription.filter:
        print(f"Filter applied: {subscription.filter}")
except NotFound:
    print("ERROR: Subscription does not exist!")

# Check if subscription topic matches
if subscription.topic != topic_path:
    print("ERROR: Subscription is attached to a different topic!")

Common Causes and Fixes

• Subscription attached to wrong topic — Verify the subscription's topic path matches your publishing target. This happens frequently when using similar topic names across environments.
• Message filters excluding messages — If your subscription has a filter, ensure published messages include the attributes the filter expects. Filters that reference non-existent attributes will silently drop all messages.
• IAM permission gaps — The publisher service account needs roles/pubsub.publisher, and the subscriber needs roles/pubsub.subscriber. Missing permissions cause silent failures in some client libraries.
• Dead letter topics misconfigured — If a dead letter policy is set but the dead letter topic doesn't exist or lacks permissions, messages may not be delivered and errors can be hard to spot.

Here's how to publish a message with attributes to ensure filters match:

import json
from google.cloud import pubsub_v1

publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path("your-project-id", "your-topic-id")

message_data = {
    "event_type": "order_created",
    "order_id": "ORD-12345",
    "amount": 99.99
}

# Encode the message
data = json.dumps(message_data).encode("utf-8")

# Include attributes that filters can match on
future = publisher.publish(
    topic_path,
    data,
    event_type="order_created",
    region="us-east1",
    priority="high"
)

# Always check the result to catch publishing failures
try:
    message_id = future.result(timeout=30)
    print(f"Published message ID: {message_id}")
except Exception as e:
    print(f"Failed to publish: {e}")

Common Issue 2: Duplicate Message Processing

Pub/Sub systems generally provide at-least-once delivery guarantees, meaning messages may be delivered more than once. This is by design — the system prioritizes delivery reliability over exactly-once semantics. If your subscriber isn't idempotent, duplicates can cause serious problems like double billing or duplicate database records.

Implementing Idempotent Consumers

The most robust solution is to make your message processing idempotent. This means processing the same message multiple times produces the same result as processing it once. The standard approach is to track processed message IDs in a durable store.

import redis
import json
from google.cloud import pubsub_v1

# Redis client for dedup tracking
redis_client = redis.Redis(host='localhost', port=6379, db=0)

subscriber = pubsub_v1.SubscriberClient()
subscription_path = subscriber.subscription_path(
    "your-project-id", "your-subscription-id"
)

def process_order(order_data):
    # Your business logic here
    print(f"Processing order: {order_data['order_id']}")
    # ... save to database, send emails, etc.

def callback(message):
    try:
        order_data = json.loads(message.data.decode("utf-8"))
        message_id = message.message_id

        # Check if we've already processed this message
        # Use SET NX (set if not exists) for atomic dedup check
        dedup_key = f"processed:{message_id}"
        was_set = redis_client.set(dedup_key, "1", nx=True, ex=86400)

        if was_set:
            # First time seeing this message — process it
            process_order(order_data)
            message.ack()
        else:
            # Already processed — just acknowledge
            print(f"Skipping duplicate message: {message_id}")
            message.ack()

    except Exception as e:
        print(f"Error processing message: {e}")
        # Nack the message so it gets redelivered
        message.nack()

# Start listening
streaming_pull = subscriber.subscribe(
    subscription_path, callback=callback
)

print(f"Listening for messages on {subscription_path}...")
try:
    streaming_pull.result()
except KeyboardInterrupt:
    streaming_pull.cancel()

Database-Level Deduplication

For systems where Redis isn't available, you can enforce deduplication at the database level using unique constraints:

import sqlite3
import json

def process_with_db_dedup(message_data, message_id):
    conn = sqlite3.connect("orders.db")
    cursor = conn.cursor()

    # Create table with unique constraint on message_id
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS processed_messages (
            message_id TEXT PRIMARY KEY,
            order_id TEXT,
            processed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    """)

    try:
        # Insert will fail if message_id already exists
        cursor.execute(
            "INSERT INTO processed_messages (message_id, order_id) VALUES (?, ?)",
            (message_id, message_data["order_id"])
        )

        # Only reaches here if insert succeeded (new message)
        cursor.execute(
            "INSERT INTO orders (order_id, amount) VALUES (?, ?)",
            (message_data["order_id"], message_data["amount"])
        )
        conn.commit()
        print(f"Processed new order: {message_data['order_id']}")

    except sqlite3.IntegrityError:
        # message_id already exists — duplicate, skip processing
        print(f"Duplicate message skipped: {message_id}")
        conn.rollback()

    finally:
        conn.close()

Common Issue 3: Message Backlog and Lag

When subscribers can't keep up with the rate of published messages, a backlog forms. This manifests as increasing oldest_unacked_message_age and growing unacked_message_count metrics. Left unchecked, backlogs can grow to the point where messages exceed retention limits and are silently dropped.

Identifying Backlog Issues

Monitor these key metrics regularly:

• oldest_unacked_message_age — how long the oldest unacked message has been waiting. Alert if this exceeds your SLA threshold.
• unacked_message_count — number of messages awaiting processing. Sudden spikes indicate subscriber problems.
• ack_message_count vs pull_request_count — if pulls are happening but acks are low, processing is failing.

Scaling Subscribers Horizontally

The most common fix for backlog is increasing subscriber parallelism. In Google Cloud Pub/Sub, you can control this through flow control settings and by running multiple subscriber instances:

from google.cloud import pubsub_v1
from concurrent.futures import ThreadPoolExecutor

subscriber = pubsub_v1.SubscriberClient()
subscription_path = subscriber.subscription_path(
    "your-project-id", "your-subscription-id"
)

# Configure flow control to maximize throughput
flow_control = pubsub_v1.types.FlowControl(
    max_messages=1000,          # Max messages held in memory
    max_bytes=100 * 1024 * 1024, # 100MB max in memory
    max_lease_duration=600,      # 10 minutes max lease
)

# Use a thread pool for parallel processing
executor = ThreadPoolExecutor(max_workers=20)

def callback(message):
    try:
        # Process the message
        process_message(message)
        message.ack()
    except Exception as e:
        print(f"Processing failed: {e}")
        message.nack()

# Start multiple streaming pulls for higher throughput
streaming_pull_futures = []
for i in range(5):
    future = subscriber.subscribe(
        subscription_path,
        callback=callback,
        flow_control=flow_control,
        executor=executor,
        await_callbacks_on_shutdown=True,
    )
    streaming_pull_futures.append(future)

print(f"Started 5 streaming pull instances")

try:
    for future in streaming_pull_futures:
        future.result()
except KeyboardInterrupt:
    for future in streaming_pull_futures:
        future.cancel()

Handling Poison Messages

A single message that consistently causes processing failures can block an entire subscription if the subscriber keeps nacking it. This creates an infinite retry loop. Implement a dead letter queue (DLQ) to isolate poison messages:

from google.cloud import pubsub_v1
import json

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

subscription_path = subscriber.subscription_path(
    "your-project-id", "your-subscription-id"
)
dlq_topic_path = publisher.topic_path(
    "your-project-id", "your-dlq-topic"
)

MAX_RETRIES = 5

def callback(message):
    try:
        data = json.loads(message.data.decode("utf-8"))
        process_message(data)

        # Clear retry count on success
        message.ack()

    except Exception as e:
        # Check delivery attempt count
        retry_count = 0
        if hasattr(message, 'delivery_attempt') and message.delivery_attempt:
            retry_count = message.delivery_attempt - 1

        print(f"Processing failed (attempt {retry_count + 1}): {e}")

        if retry_count >= MAX_RETRIES:
            # Send to dead letter queue
            print(f"Moving message to DLQ after {retry_count} attempts")
            future = publisher.publish(
                dlq_topic_path,
                message.data,
                original_message_id=message.message_id,
                error=str(e),
                retry_count=str(retry_count)
            )
            future.result(timeout=30)
            message.ack()  # Ack to remove from main subscription
        else:
            message.nack()

streaming_pull = subscriber.subscribe(subscription_path, callback=callback)
print("Listening with DLQ handling enabled...")
streaming_pull.result()

Common Issue 4: Acknowledgment Deadline Timeouts

Every Pub/Sub system has an acknowledgment deadline — the maximum time a subscriber has to process and acknowledge a message before it gets redelivered. In Google Cloud Pub/Sub, the default is 10 seconds. If your processing takes longer than this, messages will be redelivered while the original is still being processed, leading to duplicates and wasted work.

Extending Ack Deadlines

You can configure the ack deadline at the subscription level or modify it per message. For long-running processing, use the modify_ack_deadline approach:

from google.cloud import pubsub_v1
import threading
import time

subscriber = pubsub_v1.SubscriberClient()
subscription_path = subscriber.subscription_path(
    "your-project-id", "your-subscription-id"
)

# Set a longer ack deadline at subscription creation
subscription = subscriber.create_subscription(
    request={
        "name": subscription_path,
        "topic": "projects/your-project-id/topics/your-topic",
        "ack_deadline_seconds": 60,  # 60 seconds instead of default 10
    }
)

def process_long_running(message):
    """Simulate long-running processing with heartbeat."""
    def heartbeat():
        """Extend ack deadline periodically while processing."""
        while not processing_done.is_set():
            try:
                subscriber.modify_ack_deadline(
                    request={
                        "subscription": subscription_path,
                        "ack_ids": [message.ack_id],
                        "ack_deadline_seconds": 60,
                    }
                )
                print(f"Heartbeat sent for message {message.message_id}")
            except Exception as e:
                print(f"Heartbeat failed: {e}")
                break
            time.sleep(45)  # Send heartbeat before deadline expires

    processing_done = threading.Event()
    heartbeat_thread = threading.Thread(target=heartbeat, daemon=True)
    heartbeat_thread.start()

    try:
        # Your long-running processing here
        time.sleep(120)  # Simulating 2-minute processing
        print(f"Processing complete for {message.message_id}")
        message.ack()
    except Exception as e:
        print(f"Processing failed: {e}")
        message.nack()
    finally:
        processing_done.set()
        heartbeat_thread.join(timeout=5)

def callback(message):
    process_long_running(message)

streaming_pull = subscriber.subscribe(subscription_path, callback=callback)
print("Listening with heartbeat-based ack extension...")
streaming_pull.result()

Common Issue 5: Connection and Network Issues

Subscriber connections can drop due to network partitions, DNS failures, or service disruptions. Without proper handling, your subscriber may silently stop receiving messages while appearing healthy to monitoring systems.

Implementing Robust Reconnection Logic

from google.cloud import pubsub_v1
import time
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

subscriber = pubsub_v1.SubscriberClient()
subscription_path = subscriber.subscription_path(
    "your-project-id", "your-subscription-id"
)

MAX_RECONNECT_ATTEMPTS = 10
RECONNECT_DELAY = 5  # seconds

def callback(message):
    try:
        process_message(message)
        message.ack()
    except Exception as e:
        logger.error(f"Callback error: {e}")
        message.nack()

def start_subscriber():
    """Start subscriber with automatic reconnection."""
    attempt = 0

    while attempt < MAX_RECONNECT_ATTEMPTS:
        try:
            logger.info(f"Starting subscriber (attempt {attempt + 1})")
            streaming_pull = subscriber.subscribe(
                subscription_path, callback=callback
            )

            # Block until the stream is closed or errors
            streaming_pull.result()

        except Exception as e:
            attempt += 1
            logger.error(f"Subscriber error: {e}")

            if attempt < MAX_RECONNECT_ATTEMPTS:
                delay = RECONNECT_DELAY * (2 ** attempt)  # Exponential backoff
                logger.info(f"Reconnecting in {delay} seconds...")
                time.sleep(delay)
            else:
                logger.error("Max reconnection attempts reached. Exiting.")
                raise

        else:
            # Normal shutdown
            logger.info("Subscriber shut down gracefully")
            break

if __name__ == "__main__":
    start_subscriber()

Best Practices for Reliable Pub/Sub Systems

Design for At-Least-Once Delivery

Always assume messages will be delivered more than once. Design consumers to be idempotent from the start rather than retrofitting deduplication after encountering issues in production. Use unique message identifiers or business-level keys to detect and skip duplicates.

Monitor Proactively

Set up alerts on critical metrics before they become problems. Recommended alerting thresholds include:

• oldest_unacked_message_age exceeding 5 minutes (adjust based on your SLA)
• unacked_message_count growing continuously over a 10-minute window
• Publish or subscribe error rates exceeding 1% of total operations
• Dead letter topic receiving any messages (indicates processing failures)

Use Ordering When Required

If message ordering matters, enable message ordering on your topic and subscription. Be aware that ordering keys can reduce parallelism — messages with the same key are processed sequentially. Use ordering only when truly necessary:

from google.cloud import pubsub_v1

publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path("your-project-id", "your-ordered-topic")

# Publish with an ordering key
future = publisher.publish(
    topic_path,
    b'{"event": "update", "entity_id": "user-123"}',
    ordering_key="user-123"  # Messages with same key delivered in order
)
message_id = future.result()
print(f"Published ordered message: {message_id}")

Implement Proper Error Handling in Publishers

Always check the result of publish operations. Failing to do so means you may silently lose messages:

from google.cloud import pubsub_v1
from google.api_core import retry

publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path("your-project-id", "your-topic")

# Configure retry for transient failures
retry_policy = retry.Retry(
    initial=0.1,
    maximum=60,
    multiplier=2,
    deadline=300,
    predicate=retry.if_exception_type(
        ConnectionError,
        TimeoutError,
    ),
)

def publish_with_retry(data, **attributes):
    """Publish with automatic retry and error handling."""
    try:
        future = publisher.publish(topic_path, data, **attributes)
        message_id = future.result(timeout=60)
        logger.info(f"Published message: {message_id}")
        return message_id
    except Exception as e:
        logger.error(f"Publish failed after retries: {e}")
        # Implement fallback: local queue, alternative topic, etc.
        raise

# Usage
publish_with_retry(
    b'{"event": "user_signup"}',
    event_type="signup",
    source="web-app"
)

Test Failure Scenarios

Don't wait for production to discover how your system handles failures. Regularly test:

• Subscriber crashes mid-processing — verify messages are redelivered
• Network partitions — verify reconnection logic works
• Publisher failures — verify no messages are lost
• High message volume — verify backpressure handling
• Poison messages — verify DLQ routing works

Conclusion

Troubleshooting Pub/Sub systems requires a systematic approach that combines monitoring, proper error handling, and defensive consumer design. The most common issues — undelivered messages, duplicates, backlogs, ack deadline timeouts, and connection failures — all have well-established solutions. By implementing idempotent consumers, proactive monitoring with meaningful alerts, dead letter queues for poison messages, and robust reconnection logic, you can build Pub/Sub-based systems that remain reliable even under adverse conditions. Remember that at-least-once delivery is a guarantee, not a bug — design your consumers accordingly from day one. The investment in proper error handling and observability will pay dividends the first time a production issue occurs, turning what could be hours of frantic debugging into a quick, well-understood resolution.