RabbitMQ Subscriber System

This project implements a robust RabbitMQ subscriber system with error handling, retry mechanism, and a "trash" system for failed messages. The focus is solely on the subscriber functionality, making it ideal for integration into existing systems that already have message producers.

Architecture

The system consists of the following components:

1. Subscriber: Consumes messages from RabbitMQ with error handling and retry mechanism
2. Dead Letter Queue: Stores messages that failed processing after maximum retries
3. Failed Message Repository: Database storage for messages that couldn't be processed

Docker Compose Setup

The application is containerized using Docker and can be run using Docker Compose with the following services:

1. Spring Boot Application: The main application that provides the subscriber functionality
2. RabbitMQ: Message broker with management plugin for monitoring queues and exchanges

This setup ensures that all components are properly configured and can communicate with each other without manual setup.

Features

• Message consumption with validation
• Support for different message types (TEXT, JSON)
• Define at configuration level retryable and DLQ exceptions
• Error handling for invalid messages (non-consumable)
• Retry mechanism for processing errors with exponential backoff
• Dead letter queue for messages that failed after maximum retries
• Consumer that consumes the valid messages and store them in the db

Error Handling

The system handles two types of errors:

1. Non-consumable messages: Messages that are invalid and cannot be processed (e.g., invalid format, unsupported type). These are immediately sent to the dead letter queue.

2. Processing errors: Errors that occur during message processing. The system uses a configurable approach to handle these errors:

   • Exceptions can be configured to be retried or sent directly to the dead letter queue based on their type
   • For exceptions configured for retry, the system will retry processing up to a configurable maximum number of attempts
   • If the message still fails after the maximum retries, it is sent to the dead letter queue
   • For exceptions configured for direct DLQ, the message is immediately sent to the dead letter queue without retry

The system uses reflection to determine how to handle each exception type, allowing for fine-grained control over error handling without modifying the code.

Configuration

The system can be configured through the application.properties file:

# RabbitMQ Configuration
spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

# Queue Configuration
rabbitmq.queue.name=rabbit-poc-message-queue
rabbitmq.exchange.name=rabbit-poc-message-exchange
rabbitmq.routing.key=rabbit-poc-message-routing-key
rabbitmq.dead-letter.queue=rabbit-poc-dead-letter-queue
rabbitmq.dead-letter.exchange=rabbit-poc-dead-letter-exchange
rabbitmq.dead-letter.routing-key=rabbit-poc-dead-letter-routing-key

# Retry Configuration
rabbitmq.retry.max-attempts=3
rabbitmq.retry.initial-interval=1000
rabbitmq.retry.multiplier=2.0
rabbitmq.retry.max-interval=10000

# Exception Handling Configuration
# Comma-separated list of fully qualified exception class names
rabbitmq.exceptions.retry=java.io.IOException,java.net.SocketTimeoutException,org.springframework.dao.DataAccessException
rabbitmq.exceptions.dlq=java.lang.IllegalArgumentException,java.lang.NullPointerException,com.fasterxml.jackson.core.JsonProcessingException

# Database Configuration
spring.datasource.url=jdbc:postgresql://localhost:5432/rabbit_poc
spring.datasource.username=postgres
spring.datasource.password=postgres
spring.jpa.hibernate.ddl-auto=update
spring.jpa.properties.hibernate.dialect=org.hibernate.dialect.PostgreSQLDialect

Usage

Consuming Messages

The subscriber automatically consumes messages from the configured RabbitMQ queue. To use the subscriber in your application:

1. Configure the RabbitMQ connection in application.properties
2. Ensure the queue exists in your RabbitMQ server
3. Deploy the application
4. Send messages to the queue using any RabbitMQ producer

The subscriber will process messages based on their type:

• TEXT messages: Simple text processing
• JSON messages: JSON validation and processing

Handling Failed Messages

Failed messages are automatically stored in the database. You can query the database directly to retrieve and analyze failed messages:

-- Get all failed messages
SELECT * FROM failed_messages;

-- Get failed messages by error type
SELECT * FROM failed_messages WHERE error_type = 'INVALID_FORMAT';

-- Get failed messages by time range
SELECT * FROM failed_messages WHERE failed_at BETWEEN '2023-01-01 00:00:00' AND '2023-12-31 23:59:59';

-- Get failed messages by queue
SELECT * FROM failed_messages WHERE original_queue = 'rabbit-poc-message-queue';

-- Get failed messages by retry count
SELECT * FROM failed_messages WHERE retry_count >= 3;

Testing

The project includes integration tests that demonstrate how to use the subscriber functionality. For testing purposes only, a simplified message publisher is included. You can run the tests using:

./mvnw test

The tests use an in-memory H2 database and a test RabbitMQ configuration to avoid affecting the main application. The test cases verify:

1. Processing of valid text messages
2. Processing of valid JSON messages
3. Handling of invalid JSON messages
4. Handling of messages with unsupported types

Requirements

• Java 17 or higher
• RabbitMQ server
• PostgreSQL database

Getting Started

Running with Docker Compose

The easiest way to run the application is using Docker Compose, which will start the application, RabbitMQ, and PostgreSQL in containers:

1. Build the application:

   ./mvnw clean package -DskipTests

2. Start the services:

   docker-compose up -d

3. Access the RabbitMQ management console at http://localhost:15672 (username: guest, password: guest) to publish test messages

4. To stop the services:

   docker-compose down

Running Locally

If you prefer to run the application locally:

1. Start RabbitMQ server

2. Create a PostgreSQL database named rabbit_poc

3. Update the database configuration in application.properties if needed

4. Run the application:

   ./mvnw spring-boot:run

5. Use the RabbitMQ management console or any RabbitMQ client to publish messages to the configured queue

License

This project is licensed under the MIT License - see the LICENSE file for details.