Pulling and Acknowledging Messages

In the previous lesson, you successfully built a complete publisher-subscriber messaging pipeline by learning how to create and manage subscriptions. You mastered the coordination between publisher and subscriber clients, implemented robust subscription management with the ensure_subscription pattern, and established the infrastructure needed to connect your published events to consuming applications.

However, creating subscriptions is only the setup phase of message consumption. Your subscription acts as a delivery route that collects messages from your topic, but you still need to actively retrieve and process those messages. This is where message pulling becomes essential. Think of your subscription as a mailbox that receives your event messages — you still need to check the mailbox, read the messages, and confirm that you've processed them successfully.

The pull-based consumption model gives you complete control over when and how you process messages. Unlike streaming approaches that push messages to your application continuously, pulling allows you to retrieve messages on demand, process them at your own pace, and handle batches of messages efficiently.

The final piece of this puzzle is acknowledgment — the mechanism that tells Pub/Sub you've successfully processed a message and it can be safely removed from the subscription. Without proper acknowledgment, messages remain in your subscription and will be redelivered, potentially causing duplicate processing or message accumulation.

In this lesson, you'll complete your event-driven messaging system by learning how to pull messages from your subscriptions and acknowledge them properly. By the end, you'll have a fully functional system that can publish events, manage subscriptions, and process messages end-to-end.

The pull() method is your primary tool for retrieving messages from subscriptions synchronously. This method allows you to fetch messages on demand, giving you control over when your application processes events:

def pull_once(max_messages=10, timeout=5):
    sub = pubsub_v1.SubscriberClient()
    resp = sub.pull(subscription=sp, max_messages=max_messages, timeout=timeout)

The max_messages parameter controls how many messages you want to retrieve in a single pull operation. Setting this to 10 means you'll get up to 10 messages at once, allowing for efficient batch processing. Starting with a single message (max_messages=1) might seem safer, but it's inefficient for production workloads where you want to process messages in batches.

The timeout parameter specifies how long the pull operation should wait for messages if none are immediately available. Setting this to 5 seconds prevents your application from busy-waiting and allows for responsive message processing. Without a timeout parameter, the pull operation might return immediately with no messages, requiring you to implement your own polling logic.

The pull response contains a received_messages attribute that holds an array of message objects:

resp = sub.pull(subscription=sp, max_messages=max_messages, timeout=timeout)
print(f"Received {len(resp.received_messages)} messages")

When you run this against a subscription with messages, you might see:

Received 3 messages

If no messages are available, the length will be 0 — this is normal behavior and doesn't indicate an error.

Each message retrieved through the pull operation has a specific structure that separates the actual event data from the delivery metadata. Understanding this structure is crucial for both processing messages and implementing proper acknowledgment.

The received message objects contain two key components: the message attribute that holds your published data, and the ack_id attribute that provides the acknowledgment identifier:

for m in resp.received_messages:
    # m.message contains your published data
    # m.ack_id contains the acknowledgment identifier

The acknowledgment ID (ack_id) is a unique identifier that Pub/Sub generates for each message delivery. This ID serves as a receipt that proves you received a particular message and allows you to confirm successful processing later. Importantly, the acknowledgment ID is attached directly to the received message object, not nested within the message data itself.

When collecting acknowledgment IDs for batch processing, you access them directly:

ack_ids = [m.ack_id for m in resp.received_messages]

Understanding this structure prevents common mistakes like trying to access m.message.ack_id (which doesn't exist) or forgetting to collect acknowledgment IDs before processing begins.

The actual message content is stored in the message.data attribute as bytes. Since you published JSON-encoded user signup events in previous lessons, you'll need to decode and parse this data:

for m in resp.received_messages:
    message_data = json.loads(m.message.data)
    print("handling", message_data)

When processing your user signup events, you might see output like:

handling {'user_id': 'a1b2c3d4e5f6', 'email': 'user@example.com', 'timestamp': '2024-06-01T12:00:00Z', 'plan': 'free'}

Message processing can fail for various reasons — malformed JSON, missing required fields, network issues, or business logic errors. When processing fails, you need to handle these errors gracefully:

processed_ack_ids = []
for m in resp.received_messages:
    try:
        message_data = json.loads(m.message.data)
        print("handling", message_data)
        # Only add to processed list if processing succeeds
        processed_ack_ids.append(m.ack_id)
    except Exception as e:
        print(f"Failed to process message: {e}")
        # Don't add to processed_ack_ids - message will be redelivered

This selective acknowledgment approach ensures that only successfully processed messages are marked for acknowledgment, while failed messages remain unacknowledged and will be redelivered for retry processing.

Acknowledgment is the critical final step that tells Pub/Sub you've successfully handled a message and it can be safely removed from your subscription:

if processed_ack_ids:
    sub.acknowledge(subscription=sp, ack_ids=processed_ack_ids)

The conditional check ensures you only attempt acknowledgment when you actually have messages to acknowledge. The acknowledge() method processes acknowledgment IDs as a batch for efficiency — this is more performant than acknowledging messages individually.

Timing is crucial: You should only acknowledge messages after you've successfully completed all processing logic. Acknowledgment is permanent — once you acknowledge a message, it's removed from the subscription and cannot be retrieved again.

Production applications typically need to process messages continuously until a subscription is empty. This requires implementing pull loops that handle varying message volumes efficiently:

def pull_all_messages():
    total_processed = 0
    while True:
        sub = pubsub_v1.SubscriberClient()
        resp = sub.pull(subscription=sp, max_messages=10, timeout=5)
        
        if not resp.received_messages:
            break  # No more messages available
            
        print(f"Received {len(resp.received_messages)} messages")
        
        # Process messages with error handling
        processed_ack_ids = []
        for m in resp.received_messages:
            try:
                message_data = json.loads(m.message.data)
                print("handling", message_data)
                processed_ack_ids.append(m.ack_id)
                total_processed += 1
            except Exception as e:
                print(f"Failed to process message: {e}")
        
        if processed_ack_ids:
            sub.acknowledge(subscription=sp, ack_ids=processed_ack_ids)
    
    return total_processed

This pattern continues pulling until the subscription is empty and tracks the total number of successfully processed messages across all pull operations.

Here's how all components work together with batch publishing and continuous processing:

import os, json, uuid
from google.cloud import pubsub_v1
from google.api_core.exceptions import NotFound

os.environ.setdefault("PUBSUB_EMULATOR_HOST", "localhost:8681")
print("Pub/Sub emulator target →", os.environ["PUBSUB_EMULATOR_HOST"])

PROJECT, TOPIC, SUB = "demo-local", "user-signups", "signup-processor"
publisher = pubsub_v1.PublisherClient()
subscriber = pubsub_v1.SubscriberClient()
tp = publisher.topic_path(PROJECT, TOPIC)
sp = subscriber.subscription_path(PROJECT, SUB)

# Ensure topic exists
try:
    publisher.create_topic(name=tp)
except Exception:
    pass

# Ensure subscription exists
def ensure_subscription():
    try:
        subscriber.get_subscription(subscription=sp)
    except NotFound:
        subscriber.create_subscription(name=sp, topic=tp)
    return sp

def publish_multiple_signups():
    message_ids = []
    for i in range(5):
        signup = {
            "user_id": uuid.uuid4().hex,
            "email": f"user{i+1}@example.com",
            "timestamp": "2024-06-01T12:00:00Z",
            "plan": "free"
        }
        message_id = publisher.publish(tp, data=json.dumps(signup).encode()).result()
        message_ids.append(message_id)
    return message_ids

if __name__ == "__main__":
    ensure_subscription()
    message_ids = publish_multiple_signups()
    print(f"Published {len(message_ids)} messages")
    total_processed = pull_all_messages()
    print(f"Total messages processed: {total_processed}")

When you run this complete script, you'll see output like:

Pub/Sub emulator target → localhost:8681
Published 5 messages
Received 5 messages
handling {'user_id': 'a1b2c3d4e5f6', 'email': 'user1@example.com', 'timestamp': '2024-06-01T12:00:00Z', 'plan': 'free'}
handling {'user_id': 'b2c3d4e5f6a7', 'email': 'user2@example.com', 'timestamp': '2024-06-01T12:00:00Z', 'plan': 'free'}
...
Total messages processed: 5

You've successfully mastered the complete message processing cycle! You now understand how to configure pull operations with proper batching and timeouts, extract and process message data with robust error handling, implement selective acknowledgment that prevents message loss, and build continuous processing loops for production workloads.

Your event-driven messaging system now supports the complete workflow from event publishing through consumption and acknowledgment. This foundation enables reliable message processing patterns where you can control timing, handle errors gracefully, and ensure proper message cleanup.

In the upcoming practice exercises, you'll get hands-on experience with pull method configuration, message processing loops, acknowledgment ID handling, error recovery with selective acknowledgment, and continuous processing systems. These exercises will reinforce the concepts you've learned and prepare you for building production-ready message processing applications.