Message Processing with Failure Handling

Build a robust message processing system that handles messages from a queue with automatic retry and failure handling capabilities. Your system should process messages reliably and automatically route messages that repeatedly fail to a separate failure handling mechanism.

Requirements

Core Functionality

1. Message Queue Setup: Create a primary message topic and subscription for incoming messages
2. Dead Letter Configuration: Configure the subscription to automatically redirect messages that fail processing multiple times to a dedicated failure handling topic
3. Message Processing: Implement message consumption that acknowledges successfully processed messages
4. Failure Handling: Set up a separate subscription to monitor and retrieve messages that failed processing

Specific Requirements

• The primary subscription should be configured to send messages to the dead letter topic after a specified number of delivery attempts (e.g., 5 attempts)
• Successfully processed messages must be properly acknowledged to prevent redelivery
• Failed messages should automatically be moved to the dead letter topic without manual intervention
• The system should allow monitoring of failed messages through the dead letter subscription

Test Cases

• Create a topic and subscription with dead letter configuration, publish a test message, successfully pull and acknowledge it, and verify it is not redelivered @test
• Configure a subscription with dead letter topic (3 delivery attempts), publish a message, repeatedly pull without acknowledging to simulate failures, and verify the message appears in the dead letter subscription after max attempts @test

Implementation

@generates

API

class MessageProcessor:
    """Handles message processing with dead letter queue support."""

    def __init__(self, project_id, primary_topic_name, dead_letter_topic_name):
        """Initialize the message processor with project and topic names."""
        pass

    def setup_infrastructure(self, max_delivery_attempts=5):
        """
        Create topics and configure subscription with dead letter handling.

        Args:
            max_delivery_attempts: Number of delivery attempts before moving to dead letter queue
        """
        pass

    def process_message(self, message_data):
        """
        Process a single message from the primary subscription.

        Returns:
            True if message was processed successfully, False otherwise
        """
        pass

    def get_dead_letter_messages(self, max_messages=10):
        """
        Retrieve messages from the dead letter subscription.

        Returns:
            List of messages that failed processing
        """
        pass

Dependencies { .dependencies }

gcloud { .dependency }

Provides Google Cloud Pub/Sub messaging capabilities with dead letter queue support.