Chain of Responsibility Pipeline Design Pattern Implementation

Comprehensive low-level design for implementing a flexible, extensible Chain of Responsibility pattern with dynamic middleware pipelines and error handling.

PublishedJune 7, 2026
Length7,812 words
Reading time40 min read

Backend LLD

The Chain of Responsibility pattern provides a powerful mechanism for processing requests through a sequence of handlers, enabling flexible and extensible backend systems where each handler can either process the request or pass it along to the next component in the chain.

1. Problem Statement & Requirements

Clarifying questions

What is the scope of the "Chain of Responsibility" pattern in this system? Is it for request processing, validation, logging, or some other specific use case?
What is the expected scale of this pipeline? Should we design for millions of concurrent requests or a more modest throughput?
What are the key entities and their relationships? Are we processing specific business objects like orders, users, or generic payloads?
What is the read/write ratio? Is this primarily a write-heavy pipeline, or does it handle significant read operations?
What are the latency sensitivity requirements? Should we optimize for p99 latencies under 100ms or can we tolerate higher latencies?
What authentication and authorization model is in place? Will each pipeline stage require specific permissions or can we assume a trusted context?
What external systems or services will this pipeline integrate with? Are there any specific integration patterns or protocols (e.g., REST, gRPC) that need to be considered?

Functional requirements

Support a configurable chain of responsibility where each stage can process, filter, or transform input data
Allow stages to be dynamically added, removed, or reordered without system downtime
Enable each stage in the pipeline to either continue processing to the next stage or terminate the chain early
Support asynchronous execution of stages where appropriate (e.g., logging, external service calls)
Provide a standardized way to handle errors and propagate them through the pipeline
Support logging and monitoring of each stage's execution for observability

Non-functional requirements

Latency targets: p50 < 20ms, p95 < 50ms, p99 < 100ms for synchronous stages
Throughput: Support 50,000+ requests per second for the pipeline
Availability: 99.9% uptime with graceful degradation capability
Scalability: Support horizontal scaling to handle traffic spikes
Reliability: At-least-once delivery for asynchronous stages, idempotent processing where possible

Constraints and assumptions

The pipeline will be used primarily for request processing in a microservices architecture
Each stage in the chain will be implemented as a separate service or component
The pipeline will support both synchronous and asynchronous stages
All communication between stages will be via message queues or HTTP/REST
We assume a trusted environment where internal service communication is secure
The system will be deployed in a Kubernetes cluster with horizontal scaling capabilities

Out of scope

Authentication and authorization for individual pipeline stages (assumed to be handled at the service boundary)
Full end-to-end request tracing across microservices (assumed to be handled by distributed tracing tools)
Complex state management or persistence within the pipeline stages (assumed to be handled by external systems)
Pipeline configuration UI or management console (assumed to be handled by separate admin services)
Integration with external systems beyond basic HTTP/gRPC calls (assumed to be handled by specific adapter services)

2. API Design

API Style Selection

For this Chain of Responsibility pipeline design, I'll select gRPC as the API style. This choice is driven by:

Strong typing and schema definition: gRPC's Protocol Buffers provide clear, versioned contracts that are essential for pipeline components to understand data flow
Efficient binary serialization: Better performance characteristics than JSON for internal service communication
Built-in support for streaming: Enables real-time pipeline processing and async event handling
First-class error handling: Native support for structured error responses that align well with pipeline failure propagation
Service contracts: Clear separation between pipeline stages and their expected inputs/outputs

Endpoint Definitions

Method	Path	Description	Auth	Rate Limit
ProcessRequest	`/pipeline.v1.PipelineService/ProcessRequest`	Main entry point for pipeline execution	JWT	1000 req/min
GetPipelineStatus	`/pipeline.v1.PipelineService/GetPipelineStatus`	Query pipeline execution status	JWT	500 req/min
CancelPipeline	`/pipeline.v1.PipelineService/CancelPipeline`	Cancel ongoing pipeline execution	JWT	100 req/min
ListPipelines	`/pipeline.v1.PipelineService/ListPipelines`	Get list of recent pipeline executions	JWT	200 req/min

Request and Response Schemas

Request Schema (Protobuf):

message PipelineRequest {
  string id = 1;
  string payload = 2;
  map<string, string> metadata = 3;
  repeated PipelineStage stages = 4;
  int64 timeout_ms = 5;
}

message PipelineStage {
  string name = 1;
  string handler_type = 2;
  map<string, string> config = 3;
  int32 order = 4;
}

Response Schema (Protobuf):

message PipelineResponse {
  string id = 1;
  string status = 2; // "SUCCESS", "FAILED", "CANCELLED"
  string error_message = 3;
  repeated StageResult stage_results = 4;
  int64 completion_time_ms = 5;
}

message StageResult {
  string stage_name = 1;
  string status = 2; // "PENDING", "PROCESSING", "SUCCESS", "FAILED"
  string error_message = 3;
  int64 duration_ms = 4;
}

HTTP Status Codes and Error Format

Standard gRPC error codes will be used:

OK (0): Success
INVALID_ARGUMENT (3): Invalid pipeline configuration or request format
DEADLINE_EXCEEDED (4): Pipeline execution timeout
UNAUTHENTICATED (16): Authentication failure
PERMISSION_DENIED (7): Authorization failure
INTERNAL (13): Internal pipeline processing error

Standardized error response format:

message ErrorResponse {
  string error_code = 1;
  string message = 2;
  repeated string details = 3;
  int64 timestamp = 4;
}

API Versioning Strategy

Versioning will be handled through:

gRPC package versioning: pipeline.v1, pipeline.v2 for major changes
Backward compatibility: New fields will be optional with default values
Feature flags: For gradual rollout of new pipeline stages

Pagination Design

For list operations, cursor-based pagination will be used:

Reasoning: More stable and consistent than offset-based pagination, especially when data is being inserted concurrently
Implementation: Cursor will be a timestamp-based token to ensure consistent ordering

Authentication and Authorization

Authentication: JWT tokens with standard OAuth2 flow for service-to-service authentication.

Authorization:

RBAC-based: Users must have pipeline.execute permission to process requests
Pipeline-specific roles: pipeline.admin for management operations, pipeline.executor for execution

Rate Limiting Strategy

Token bucket algorithm: 1000 tokens per minute for main processing endpoint
Per-user rate limiting: Each authenticated user gets 1000 tokens/min
IP-based fallback: For unauthenticated requests, rate limit by IP address (100 req/min)
Enforcement: Middleware layer intercepts and enforces limits before request processing

Idempotency Design

For mutation endpoints:

Idempotency Key: X-Idempotency-Key header containing UUID
Storage Strategy: Store request metadata with key → pipeline ID mapping for 5 minutes
Duplicate Detection: Check if same idempotency key was used within last 5 minutes
Behavior: Return same pipeline ID and status for duplicate requests

// Example idempotency check logic (simplified)
public class IdempotencyService {
    private final Map<String, String> idempotencyCache = new ConcurrentHashMap<>();
    
    public Optional<String> checkIdempotency(String idempotencyKey) {
        String existingPipelineId = idempotencyCache.get(idempotencyKey);
        if (existingPipelineId != null) {
            return Optional.of(existingPipelineId);
        }
        return Optional.empty();
    }
    
    public void storeIdempotency(String idempotencyKey, String pipelineId) {
        idempotencyCache.put(idempotencyKey, pipelineId);
        // Schedule cleanup after 5 minutes
        ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
        scheduler.schedule(() -> idempotencyCache.remove(idempotencyKey), 5, TimeUnit.MINUTES);
    }
}

3. Data Model Design

Entity-Relationship diagram (Mermaid)

erDiagram
    PIPELINE {
        string id PK
        string name
        string description
        int priority
        timestamp created_at
        timestamp updated_at
    }

    PIPELINE_ITEM {
        string id PK
        string pipeline_id FK
        string handler_type
        string config
        int order_index
        boolean is_active
        timestamp created_at
        timestamp updated_at
    }

    PIPELINE_ITEM_EXECUTION {
        string id PK
        string pipeline_item_id FK
        string input_data
        string output_data
        string status
        timestamp started_at
        timestamp completed_at
        string error_message
        int execution_duration_ms
    }

    PIPELINE_ITEM ||--o{ PIPELINE_ITEM_EXECUTION : "has"
    PIPELINE ||--o{ PIPELINE_ITEM : "contains"

Table / Collection schema definitions with field types, constraints, indices

Pipeline table

CREATE TABLE pipelines (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    name VARCHAR(255) NOT NULL,
    description TEXT,
    priority INTEGER NOT NULL DEFAULT 0,
    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
    updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
    
    CONSTRAINT chk_priority_positive CHECK (priority >= 0)
);

CREATE INDEX idx_pipelines_priority ON pipelines(priority);
CREATE INDEX idx_pipelines_created_at ON pipelines(created_at);

PipelineItem table

CREATE TABLE pipeline_items (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    pipeline_id UUID NOT NULL REFERENCES pipelines(id) ON DELETE CASCADE,
    handler_type VARCHAR(255) NOT NULL,
    config JSONB DEFAULT '{}',
    order_index INTEGER NOT NULL,
    is_active BOOLEAN DEFAULT TRUE,
    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
    updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
    
    CONSTRAINT chk_order_index_positive CHECK (order_index >= 0),
    CONSTRAINT chk_handler_type_not_empty CHECK (LENGTH(handler_type) > 0)
);

CREATE INDEX idx_pipeline_items_pipeline_id ON pipeline_items(pipeline_id);
CREATE INDEX idx_pipeline_items_order_index ON pipeline_items(order_index);
CREATE INDEX idx_pipeline_items_active ON pipeline_items(is_active);
CREATE UNIQUE INDEX idx_pipeline_items_unique_order_per_pipeline ON pipeline_items(pipeline_id, order_index);

PipelineItemExecution table

CREATE TABLE pipeline_item_executions (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    pipeline_item_id UUID NOT NULL REFERENCES pipeline_items(id) ON DELETE CASCADE,
    input_data JSONB NOT NULL,
    output_data JSONB,
    status VARCHAR(50) NOT NULL DEFAULT 'PENDING',
    started_at TIMESTAMP WITH TIME ZONE,
    completed_at TIMESTAMP WITH TIME ZONE,
    error_message TEXT,
    execution_duration_ms INTEGER,
    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
    
    CONSTRAINT chk_status_valid CHECK (status IN ('PENDING', 'PROCESSING', 'SUCCESS', 'FAILED'))
);

CREATE INDEX idx_pipeline_item_executions_pipeline_item_id ON pipeline_item_executions(pipeline_item_id);
CREATE INDEX idx_pipeline_item_executions_status ON pipeline_item_executions(status);
CREATE INDEX idx_pipeline_item_executions_started_at ON pipeline_item_executions(started_at);

Primary key strategy (UUID vs. auto-increment vs. ULID — with justification)

Using UUIDs for all primary keys to support:

Distributed system compatibility (no coordination needed between services)
Better sharding capabilities across multiple instances
Immutable identifiers that don't change during data migration or replication
Avoiding potential hotspots in auto-increment scenarios

Index design (composite, partial, covering indices — with justification per index)

idx_pipelines_priority:
- Purpose: Efficient pipeline selection by priority
- Justification: Used for sorting and filtering pipelines in the execution engine
idx_pipelines_created_at:
- Purpose: For audit and monitoring queries
- Justification: Allows time-based filtering for pipeline lifecycle management
idx_pipeline_items_pipeline_id:
- Purpose: Fast lookup of pipeline items by pipeline
- Justification: Core relationship query for building execution chains
idx_pipeline_items_order_index:
- Purpose: Fast retrieval of items in order
- Justification: Required for maintaining execution sequence
idx_pipeline_items_active:
- Purpose: Filter active items only
- Justification: Performance optimization for runtime filtering
idx_pipeline_item_executions_pipeline_item_id:
- Purpose: Fast lookup of execution history by pipeline item
- Justification: Audit and monitoring use cases
idx_pipeline_item_executions_status:
- Purpose: Query by execution status
- Justification: For monitoring and retry logic

Data partitioning / sharding strategy

Partition by pipeline_id hash to distribute pipelines across shards. Each shard would contain a subset of pipelines and their associated items, enabling horizontal scaling.

Soft delete vs. hard delete decision with justification

Soft delete is used for pipelines and pipeline_items to:

Preserve historical execution data in pipeline_item_executions
Allow for recovery of deleted configurations
Support audit trails and compliance requirements

Audit trail design (who, what, when)

All data changes are audited via:

created_at and updated_at timestamps (system-generated)
No explicit user tracking in this model (assumes authentication layer handles that)

Data archival strategy and retention policy

Active pipelines: Retained indefinitely
Pipeline item executions:
- Keep last 30 days of successful executions
- Keep all failed executions indefinitely for debugging
- Archive to cold storage after 90 days

-- Example retention policy trigger (simplified)
CREATE OR REPLACE FUNCTION archive_old_executions()
RETURNS TRIGGER AS $$
BEGIN
    IF OLD.completed_at < NOW() - INTERVAL '90 days' THEN
        -- Move to archive table or mark as archived
        RETURN OLD;
    END IF;
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

4. Class Design & Service Layer

UML Class Diagram

classDiagram
    class Handler {
        <<Interface>>
        +handle(Request)
        +setNext(Handler)
    }
    
    class AbstractHandler {
        -next: Handler
        +setNext(Handler)
        +handle(Request)
    }
    
    class AuthenticationHandler {
        +handle(Request)
    }
    
    class AuthorizationHandler {
        +handle(Request)
    }
    
    class ValidationHandler {
        +handle(Request)
    }
    
    class LoggingHandler {
        +handle(Request)
    }
    
    class Request {
        -data: String
        -userId: String
        -action: String
    }
    
    Handler <|-- AbstractHandler
    AbstractHandler <|-- AuthenticationHandler
    AbstractHandler <|-- AuthorizationHandler
    AbstractHandler <|-- ValidationHandler
    AbstractHandler <|-- LoggingHandler
    
    class Pipeline {
        -handlers: List~Handler~
        +execute(Request)
    }
    
    Pipeline --> Handler : uses

State Machine Diagram

stateDiagram-v2
    [*] --> Ready
    Ready --> Processing : Request received
    Processing --> Validated : Validation successful
    Processing --> Failed : Validation failed
    Validated --> Completed : All handlers processed
    Validated --> Failed : Handler error
    Failed --> Retry : Retryable error
    Retry --> Processing : Retry attempt
    Completed --> [*]
    Failed --> [*]

Service Class Structure and Responsibilities

The Chain of Responsibility pipeline consists of:

Handler Interface: Defines the contract for all handlers
AbstractHandler: Base implementation with next handler chaining
Concrete Handlers: Specific implementations for different responsibilities
Pipeline Service: Orchestrates the execution flow

Interface / Abstract Class Design

public interface Handler {
    Handler setNext(Handler handler);
    boolean handle(Request request);
}

public abstract class AbstractHandler implements Handler {
    protected Handler next;
    
    @Override
    public Handler setNext(Handler handler) {
        this.next = handler;
        return handler;
    }
    
    @Override
    public boolean handle(Request request) {
        // Default implementation - can be overridden
        if (next != null) {
            return next.handle(request);
        }
        return true;
    }
}

Business Logic Encapsulation Patterns

Using the Chain of Responsibility pattern with Strategy for handler-specific logic:

public class AuthenticationHandler extends AbstractHandler {
    private final AuthenticationService authService;
    
    public AuthenticationHandler(AuthenticationService authService) {
        this.authService = authService;
    }
    
    @Override
    public boolean handle(Request request) {
        if (!authService.isAuthenticated(request.getUserId())) {
            throw new AuthenticationException("User not authenticated");
        }
        return super.handle(request);
    }
}

public class AuthorizationHandler extends AbstractHandler {
    private final AuthorizationService authzService;
    
    public AuthorizationHandler(AuthorizationService authzService) {
        this.authzService = authzService;
    }
    
    @Override
    public boolean handle(Request request) {
        if (!authzService.isAuthorized(request.getUserId(), request.getAction())) {
            throw new AuthorizationException("User not authorized for action");
        }
        return super.handle(request);
    }
}

Critical-Path Method Implementation

public class PipelineService {
    private final List<Handler> handlers;
    
    public PipelineService(List<Handler> handlers) {
        this.handlers = new ArrayList<>(handlers);
        buildChain();
    }
    
    public boolean execute(Request request) {
        try {
            // Validate input
            if (request == null) {
                throw new IllegalArgumentException("Request cannot be null");
            }
            
            // Execute the chain
            boolean success = true;
            for (Handler handler : handlers) {
                try {
                    if (!handler.handle(request)) {
                        success = false;
                        break;
                    }
                } catch (Exception e) {
                    // Log error but continue with other handlers
                    logger.error("Handler failed for request: {}", 
                        request, e);
                    success = false;
                    break;
                }
            }
            
            return success;
        } catch (Exception e) {
            logger.error("Pipeline execution failed for request: {}", 
                request, e);
            throw new PipelineExecutionException("Failed to execute pipeline", e);
        }
    }
    
    private void buildChain() {
        for (int i = 0; i < handlers.size() - 1; i++) {
            handlers.get(i).setNext(handlers.get(i + 1));
        }
    }
}

Transaction Boundary Design

Each handler in the pipeline operates as a single unit of work. The entire pipeline execution is treated as one atomic operation:

All handlers execute in sequence
If any handler fails, the pipeline stops and returns failure
No transaction scope is maintained across handlers (each handler manages its own transaction if needed)

Service Interface Contracts

public interface PipelineService {
    boolean execute(Request request);
}

public interface Handler {
    boolean handle(Request request);
}

Saga / Orchestration Pattern

For distributed systems, the pipeline can be extended with a saga pattern for cross-service coordination:

public class DistributedPipelineService {
    private final List<Handler> handlers;
    private final EventPublisher eventPublisher;
    
    public boolean execute(Request request) {
        List<String> executedHandlers = new ArrayList<>();
        
        try {
            for (Handler handler : handlers) {
                String handlerId = handler.getClass().getSimpleName();
                
                // Publish start event
                eventPublisher.publish(new HandlerStartEvent(handlerId, request));
                
                boolean success = handler.handle(request);
                executedHandlers.add(handlerId);
                
                if (!success) {
                    // Publish failure event and rollback
                    eventPublisher.publish(new HandlerFailureEvent(handlerId, request));
                    rollback(executedHandlers);
                    return false;
                }
                
                // Publish success event
                eventPublisher.publish(new HandlerSuccessEvent(handlerId, request));
            }
            
            return true;
        } catch (Exception e) {
            rollback(executedHandlers);
            throw new PipelineExecutionException("Pipeline failed", e);
        }
    }
    
    private void rollback(List<String> executedHandlers) {
        // Implement rollback logic for each handler
        for (String handlerId : executedHandlers) {
            eventPublisher.publish(new HandlerRollbackEvent(handlerId, request));
        }
    }
}

Domain Object / Value Object Design

public class Request {
    private final String data;
    private final String userId;
    private final String action;
    
    public Request(String data, String userId, String action) {
        this.data = data;
        this.userId = userId;
        this.action = action;
    }
    
    // Getters only - immutable
    public String getData() { return data; }
    public String getUserId() { return userId; }
    public String getAction() { return action; }
}

public class HandlerResult {
    private final boolean success;
    private final String errorMessage;
    
    public HandlerResult(boolean success, String errorMessage) {
        this.success = success;
        this.errorMessage = errorMessage;
    }
    
    public boolean isSuccess() { return success; }
    public String getErrorMessage() { return errorMessage; }
}

5. Concurrency & Thread Safety

Concurrency Model

The Chain of Responsibility pipeline operates in a multi-threaded environment where multiple concurrent requests may process through the same handler chain. Each request is handled in its own thread context, ensuring that individual request processing is isolated and does not interfere with other concurrent requests. The system uses a thread-per-request model, which is appropriate for handling I/O-bound operations typical in web services.

Race Condition Prevention Strategy

Race conditions are mitigated through:

Immutable Handler State: All handler instances maintain immutable state to prevent concurrent modification
Thread-Local Context: Each request's processing context is isolated in thread-local storage to prevent cross-request contamination
Synchronized Access: Critical sections that modify shared state (e.g., metrics counters) are protected with synchronized blocks or atomic operations
Immutable Domain Objects: Business entities processed by the pipeline are treated as immutable to prevent concurrent modification during processing

Optimistic vs. Pessimistic Locking Selection

The pipeline design does not require explicit locking mechanisms since:

Each handler processes a request in isolation
The pipeline is stateless and does not maintain shared mutable state between requests
If the pipeline needs to interact with database or external systems, those interactions are handled through proper transaction management rather than pipeline-level locking

Deadlock Prevention Strategy

Deadlock prevention is handled through:

Fixed Handler Order: Handlers are configured in a fixed, predetermined order to prevent circular dependencies
No Nested Locks: No handler attempts to acquire locks on other handlers or shared resources in a nested fashion
Timeout Configuration: External system calls (e.g., database queries) use timeouts to prevent indefinite blocking
Request Isolation: Each request operates in its own isolated context, eliminating cross-request dependency scenarios

Idempotency Implementation for Concurrent Duplicate Requests

Idempotency is handled at the application layer through:

Request Deduplication: Each request is assigned a unique identifier that can be used to detect and discard duplicate requests
Idempotency Key Pattern: The pipeline can be extended to support idempotency keys in request headers that are used to identify and prevent duplicate processing
Stateless Design: Since the pipeline is stateless, concurrent requests that process identical data will produce identical results without side effects

// Example of idempotency key handling in a pipeline processor
public class IdempotentPipelineProcessor {
    private final Set<String> processedRequests = new HashSet<>();
    private final ReadWriteLock lock = new ReentrantReadWriteLock();
    
    public PipelineResult processWithIdempotency(PipelineRequest request) {
        String idempotencyKey = request.getIdempotencyKey();
        
        // Check if this request was already processed
        boolean alreadyProcessed = lock.readLock().tryLock() && 
            processedRequests.contains(idempotencyKey);
        
        if (alreadyProcessed) {
            lock.readLock().unlock();
            return PipelineResult.fromCache(idempotencyKey);
        }
        
        lock.readLock().unlock();
        
        // Process the request
        PipelineResult result = processRequest(request);
        
        // Mark as processed (idempotency tracking)
        lock.writeLock().lock();
        try {
            processedRequests.add(idempotencyKey);
        } finally {
            lock.writeLock().unlock();
        }
        
        return result;
    }
}

6. Sequence Diagrams for Key Flows

sequenceDiagram
    participant Client
    participant PipelineHandler1
    participant PipelineHandler2
    participant PipelineHandler3
    participant FinalHandler

    Client->>PipelineHandler1: Request(data)
    PipelineHandler1->>PipelineHandler2: Request(data)
    PipelineHandler2->>PipelineHandler3: Request(data)
    PipelineHandler3->>FinalHandler: Request(data)
    FinalHandler-->>Client: Response(result)

sequenceDiagram
    participant Client
    participant PipelineHandler1
    participant PipelineHandler2
    participant ErrorHandler

    Client->>PipelineHandler1: Request(data)
    PipelineHandler1->>PipelineHandler2: Request(data)
    PipelineHandler2-->>Client: Error(response)

sequenceDiagram
    participant Client
    participant PipelineHandler1
    participant AsyncProcessor
    participant PipelineHandler2

    Client->>PipelineHandler1: Request(data)
    PipelineHandler1->>AsyncProcessor: AsyncTask(data)
    PipelineHandler1-->>Client: Response(processing)
    AsyncProcessor->>PipelineHandler2: Process(result)
    PipelineHandler2-->>Client: FinalResponse(result)

7. Caching Design

Cache Strategy Selection

For the Chain of Responsibility pipeline, we'll use cache-aside pattern as it provides the most flexibility for handling different types of data with varying access patterns. This approach allows us to control when and how cache is populated, making it ideal for pipeline stages that may have different caching requirements.

Cache Key Design

Cache keys will follow a structured namespace pattern:

{cache_namespace}:{entity_type}:{entity_id}:{pipeline_stage}:{version}

Example key structure:

pipeline:order:12345:validation:v1
pipeline:payment:67890:processing:v2

TTL Strategy

Data Type	TTL	Reason
Pipeline configuration	1 hour	Configuration changes are infrequent
Validation results	5 minutes	High accuracy needed, frequent validation
Processing state	30 seconds	Short-lived processing context
Cached responses	1 hour	Stable response data
Retry state tracking	10 minutes	Retry window for transient failures

Cache Eviction Policy

We'll implement a hybrid approach:

LRU (Least Recently Used) for frequently accessed pipeline stages
TTL-based eviction for time-sensitive data like processing state
Size-based limits with automatic cleanup of expired entries

Cache Stampede Prevention

To prevent cache stampedes:

Probabilistic early expiry: Set TTL to 1.5x actual expiration time
Distributed locks for cache population in high-concurrency scenarios
Cache warming during low-traffic periods

Distributed Cache Architecture

We'll use Redis Cluster with:

3 master nodes for data distribution
3 replica nodes for high availability
Automatic failover configuration
Cross-zone replication for fault tolerance

Cache Invalidation Strategy

We'll implement event-driven invalidation:

Write-through pattern: Cache updates happen during write operations
Explicit purge: Invalidate cache entries when pipeline configuration changes
Time-based expiration: Automatic cleanup of stale data

public class PipelineCacheInvalidator {
    private final RedisTemplate<String, Object> redisTemplate;
    
    public void invalidatePipelineCache(String pipelineId, String stage) {
        // Invalidate all entries for this pipeline and stage
        String pattern = "pipeline:" + pipelineId + ":" + stage + ":*";
        Set<String> keys = redisTemplate.keys(pattern);
        
        if (keys != null && !keys.isEmpty()) {
            redisTemplate.delete(keys);
        }
        
        // Also invalidate any cached configuration
        redisTemplate.delete("pipeline:" + pipelineId + ":config");
    }
    
    public void invalidateAllPipelineCaches(String pipelineId) {
        String pattern = "pipeline:" + pipelineId + ":*";
        Set<String> keys = redisTemplate.keys(pattern);
        
        if (keys != null && !keys.isEmpty()) {
            redisTemplate.delete(keys);
        }
    }
}

Cache Read-Through Strategy

public class PipelineCacheService {
    private final RedisTemplate<String, Object> redisTemplate;
    private final PipelineRepository pipelineRepository;
    
    public PipelineResult getCachedPipelineResult(String pipelineId, String stage) {
        String cacheKey = buildCacheKey(pipelineId, stage);
        
        // Try cache first
        PipelineResult cached = (PipelineResult) redisTemplate.opsForValue().get(cacheKey);
        
        if (cached != null) {
            return cached;
        }
        
        // Cache miss - fetch from database and populate cache
        PipelineResult result = pipelineRepository.getPipelineResult(pipelineId, stage);
        
        if (result != null) {
            // Cache with TTL
            redisTemplate.opsForValue().set(
                cacheKey, 
                result, 
                Duration.ofMinutes(5)
            );
        }
        
        return result;
    }
    
    private String buildCacheKey(String pipelineId, String stage) {
        return "pipeline:" + pipelineId + ":" + stage + ":v1";
    }
}

Cache Performance Considerations

Cache hit ratio target: 85%+ for pipeline operations
Memory allocation: 2GB dedicated to pipeline cache with LRU eviction
Read/write ratio: 80% reads, 20% writes - optimized for read-heavy pipeline processing
Cache warming: Pre-populate cache during system startup with common pipeline configurations

Cache Monitoring

We'll track:

Cache hit/miss ratio
Average cache retrieval time
Memory usage per pipeline type
Eviction rates and reasons

This caching design provides optimal performance for the Chain of Responsibility pipeline while maintaining data consistency and supporting the distributed nature of pipeline processing.

8. Repository / Data Access Layer

Repository Pattern Implementation

The repository pattern is implemented to provide a clean abstraction over data access operations, decoupling the service layer from the underlying persistence mechanism. Each entity has its own repository interface and implementation.

public interface UserRepository {
    User findById(UUID id);
    User findByEmail(String email);
    List<User> findAll();
    User save(User user);
    void delete(UUID id);
    boolean existsById(UUID id);
}

@Repository
public class UserRepositoryImpl implements UserRepository {
    private final JdbcTemplate jdbcTemplate;
    
    @Override
    public User findById(UUID id) {
        return jdbcTemplate.queryForObject(
            "SELECT * FROM users WHERE id = ?", 
            new UserRowMapper(), 
            id
        );
    }
    
    @Override
    public User findByEmail(String email) {
        return jdbcTemplate.queryForObject(
            "SELECT * FROM users WHERE email = ?", 
            new UserRowMapper(), 
            email
        );
    }
    
    @Override
    public List<User> findAll() {
        return jdbcTemplate.query(
            "SELECT * FROM users", 
            new UserRowMapper()
        );
    }
    
    @Override
    public User save(User user) {
        if (user.getId() == null) {
            user.setId(UUID.randomUUID());
            jdbcTemplate.update(
                "INSERT INTO users (id, email, name, created_at) VALUES (?, ?, ?, ?)",
                user.getId(), 
                user.getEmail(), 
                user.getName(), 
                LocalDateTime.now()
            );
        } else {
            jdbcTemplate.update(
                "UPDATE users SET email = ?, name = ?, updated_at = ? WHERE id = ?",
                user.getEmail(), 
                user.getName(), 
                LocalDateTime.now(), 
                user.getId()
            );
        }
        return user;
    }
    
    @Override
    public void delete(UUID id) {
        jdbcTemplate.update("DELETE FROM users WHERE id = ?", id);
    }
    
    @Override
    public boolean existsById(UUID id) {
        return jdbcTemplate.queryForObject(
            "SELECT COUNT(*) FROM users WHERE id = ?", 
            Integer.class, 
            id
        ) > 0;
    }
}

ORM vs Raw Query Decisions

For this design, raw SQL queries are preferred over ORM for the following reasons:

Performance: Raw queries provide better performance with reduced overhead
Control: Full control over query execution and optimization
Complex operations: Complex joins, window functions, and analytics queries are more readable in SQL
Database-specific features: Better utilization of database-specific capabilities

Read vs Write Separation

The system implements read-write separation using a CQRS pattern for high-performance scenarios:

// Read-side repository (optimized for reads)
@Repository
public class UserReadRepository {
    private final JdbcTemplate readJdbcTemplate;
    
    public List<User> findActiveUsers() {
        return readJdbcTemplate.query(
            "SELECT u.id, u.email, u.name FROM users u WHERE u.status = 'ACTIVE' ORDER BY u.created_at DESC LIMIT 100",
            new UserRowMapper()
        );
    }
}

// Write-side repository (optimized for writes)
@Repository
public class UserWriteRepository {
    private final JdbcTemplate writeJdbcTemplate;
    
    public User save(User user) {
        // Optimized for write operations with minimal overhead
        return writeJdbcTemplate.update(
            "INSERT INTO users (id, email, name, status, created_at) VALUES (?, ?, ?, ?, ?)",
            user.getId(), 
            user.getEmail(), 
            user.getName(), 
            user.getStatus().toString(), 
            LocalDateTime.now()
        ) > 0 ? user : null;
    }
}

Connection Pool Configuration

Connection pool is configured with optimal parameters for high throughput:

# application.yml
spring:
  datasource:
    hikari:
      maximum-pool-size: 20
      minimum-idle: 5
      connection-timeout: 30000
      idle-timeout: 600000
      max-lifetime: 1800000
      validation-timeout: 5000
      leak-detection-threshold: 60000

N+1 Query Prevention Strategy

The design prevents N+1 queries through eager loading and batch operations:

// Instead of N+1 queries, use batch loading
public List<User> findUsersWithOrders(List<UUID> userIds) {
    // Single query to fetch users
    List<User> users = jdbcTemplate.query(
        "SELECT * FROM users WHERE id IN (?)",
        new Object[]{userIds.toArray()},
        new UserRowMapper()
    );
    
    // Batch query to fetch orders for all users
    List<Order> orders = jdbcTemplate.query(
        "SELECT * FROM orders WHERE user_id IN (?) ORDER BY created_at DESC",
        new Object[]{userIds.toArray()},
        new OrderRowMapper()
    );
    
    // Map orders to users
    Map<UUID, List<Order>> orderMap = orders.stream()
        .collect(Collectors.groupingBy(Order::getUserId));
    
    users.forEach(user -> {
        user.setOrders(orderMap.getOrDefault(user.getId(), Collections.emptyList()));
    });
    
    return users;
}

Batch Operation Design

Batch operations are implemented for high-volume data processing:

public void batchInsertUsers(List<User> users) {
    String sql = "INSERT INTO users (id, email, name, created_at) VALUES (?, ?, ?, ?)";
    
    jdbcTemplate.batchUpdate(sql, new BatchPreparedStatementSetter() {
        @Override
        public void setValues(PreparedStatement ps, int i) throws SQLException {
            User user = users.get(i);
            ps.setObject(1, user.getId());
            ps.setString(2, user.getEmail());
            ps.setString(3, user.getName());
            ps.setTimestamp(4, Timestamp.valueOf(user.getCreatedAt()));
        }
        
        @Override
        public int getBatchSize() {
            return users.size();
        }
    });
}

Transaction Management

All repository operations are wrapped in proper transaction boundaries:

@Transactional
public User saveWithAudit(User user) {
    User savedUser = save(user);
    
    // Audit trail creation
    AuditLog auditLog = new AuditLog(
        "USER_CREATE", 
        savedUser.getId().toString(), 
        "Created user: " + savedUser.getEmail(),
        "SYSTEM"
    );
    
    auditLogRepository.save(auditLog);
    
    return savedUser;
}

Data Access Layer Interface Contracts

public interface Repository<T, ID> {
    Optional<T> findById(ID id);
    List<T> findAll();
    T save(T entity);
    void deleteById(ID id);
    boolean existsById(ID id);
}

Query Optimization

Index usage is optimized through strategic index creation:

-- User table indexes for optimal query performance
CREATE INDEX idx_users_email ON users(email);
CREATE INDEX idx_users_status_created_at ON users(status, created_at);
CREATE INDEX idx_users_created_at ON users(created_at DESC);

Read Replica Strategy

For read-heavy workloads, a read replica configuration is implemented:

// Read-only data source configuration
@Configuration
public class ReadReplicaConfig {
    @Bean
    @Primary
    public DataSource writeDataSource() {
        return HikariDataSourceBuilder.create()
            .dataSourceClassName("org.postgresql.ds.PGSimpleDataSource")
            .username("write_user")
            .password("write_password")
            .build();
    }
    
    @Bean
    public DataSource readDataSource() {
        return HikariDataSourceBuilder.create()
            .dataSourceClassName("org.postgresql.ds.PGSimpleDataSource")
            .username("read_user")
            .password("read_password")
            .build();
    }
}

Database Sharding Considerations

For future scalability, sharding strategy is designed with consistent hashing:

public class UserShardKeyGenerator {
    public String generateShardKey(UUID userId) {
        // Consistent hashing to ensure even distribution across shards
        return "shard_" + (userId.hashCode() % 32);
    }
}

Data Migration Strategy

Support for data migration through versioned repository implementations:

@Repository
public class UserMigrationRepository {
    // Migration logic for schema evolution
    public void migrateUserTable() {
        jdbcTemplate.update("ALTER TABLE users ADD COLUMN IF NOT EXISTS last_login TIMESTAMP");
        jdbcTemplate.update("UPDATE users SET last_login = created_at WHERE last_login IS NULL");
    }
}

9. Asynchronous Processing & Messaging

Event-driven boundaries in this feature — what triggers async vs. what stays synchronous, and why

The Chain of Responsibility pattern is primarily a synchronous processing pipeline where each handler in the chain processes a request sequentially. However, for scalability and decoupling, we can introduce asynchronous processing in two key areas:

Async Event Publishing: When a handler completes its processing and needs to notify downstream systems or trigger side effects (e.g., sending notifications, updating analytics), these should be published asynchronously to avoid blocking the main processing flow.
Async Handler Execution: If a handler in the chain performs long-running operations (e.g., external API calls, heavy computation), these can be offloaded to a background task queue for better responsiveness.

We will keep the core Chain of Responsibility logic synchronous but introduce async publishing of events to decouple downstream systems.

Message schema design with a sample payload (Avro / Protobuf / JSON Schema — with versioning strategy)

{
  "type": "object",
  "title": "ChainOfResponsibilityEvent",
  "version": "1.0.0",
  "properties": {
    "eventId": {
      "type": "string",
      "description": "Unique identifier for the event"
    },
    "chainId": {
      "type": "string",
      "description": "Identifier for the chain instance"
    },
    "handlerName": {
      "type": "string",
      "description": "Name of the handler that processed this event"
    },
    "payload": {
      "type": "object",
      "description": "The payload being processed"
    },
    "timestamp": {
      "type": "string",
      "format": "date-time",
      "description": "ISO 8601 timestamp"
    },
    "status": {
      "type": "string",
      "enum": ["SUCCESS", "FAILED"],
      "description": "Status of the handler execution"
    },
    "errorDetails": {
      "type": "object",
      "properties": {
        "code": { "type": "string" },
        "message": { "type": "string" }
      }
    }
  },
  "required": ["eventId", "chainId", "handlerName", "payload", "timestamp", "status"]
}

Versioning strategy: Use semantic versioning (MAJOR.MINOR.PATCH) for message schemas. Backward compatibility is maintained by ensuring that new fields are optional and not breaking existing consumers.

Dead letter queue strategy and poison message handling

Poison messages (messages that repeatedly fail processing) will be routed to a dead letter queue after a configurable number of retries (e.g., 3 attempts). The system will:

Track failed message counts per handler.
Move messages to a DLQ after exceeding retry limits.
Log the failed message details for manual inspection and reprocessing.

At-least-once vs. exactly-once delivery — decision with justification for this feature

We choose at-least-once delivery for the Chain of Responsibility pipeline because:

The pattern is designed for processing a request through a sequence of handlers, where each handler may perform side effects or state changes.
Ensuring exactly-once delivery would require complex state tracking and distributed transaction management, which adds significant complexity.
For most use cases in a Chain of Responsibility pipeline, idempotency at the handler level is sufficient to handle duplicate processing.

Retry policy for consumers (max attempts, backoff formula, non-retryable error classification)

Retry policy configuration:

Max attempts: 3
Backoff formula: Exponential backoff with jitter (Jitter = random(0, 1) × baseDelay)
Base delay: 1 second
Max retry delay: 30 seconds

Non-retryable errors:

Validation failures (400 status codes)
Authorization errors (401/403)
Business logic errors that indicate the message is malformed or invalid

Outbox pattern for reliable event publishing — include or explicitly justify omission

We will NOT use the outbox pattern in this design because:

The Chain of Responsibility is a synchronous pipeline, and we're only introducing async event publishing for side effects.
The system assumes that the messaging infrastructure (e.g., Kafka or RabbitMQ) provides at-least-once delivery guarantees.
Implementing outbox pattern adds significant complexity and overhead for a simple event publishing use case.

Instead, we rely on the messaging system's built-in reliability features and ensure that event publishing is wrapped in a transactional context where possible.

Sequence diagram for all async flows (Mermaid)

sequenceDiagram
    participant Client
    participant HandlerChain
    participant MessageBroker
    participant ConsumerA
    participant ConsumerB

    Client->>HandlerChain: Submit request
    HandlerChain->>HandlerChain: Process through handlers
    HandlerChain->>MessageBroker: Publish event (async)
    MessageBroker->>ConsumerA: Deliver event
    ConsumerA-->>MessageBroker: Acknowledge receipt
    MessageBroker->>ConsumerB: Deliver event
    ConsumerB-->>MessageBroker: Acknowledge receipt

10. Error Handling Design

Error Classification Hierarchy

The system implements a three-level error classification hierarchy to enable targeted handling and monitoring:

Domain Errors - Business logic violations (e.g., insufficient funds, invalid input)
Infrastructure Errors - System-level issues (e.g., database connection failure, timeout)
Validation Errors - Input validation failures (e.g., missing required fields, format issues)

public abstract class BaseException extends RuntimeException {
    private final String errorCode;
    private final int httpStatus;

    public BaseException(String message, String errorCode, int httpStatus) {
        super(message);
        this.errorCode = errorCode;
        this.httpStatus = httpStatus;
    }

    // Getters
}

public class DomainException extends BaseException {
    public DomainException(String message, String errorCode) {
        super(message, errorCode, 400);
    }
}

public class InfrastructureException extends BaseException {
    public InfrastructureException(String message, String errorCode) {
        super(message, errorCode, 500);
    }
}

public class ValidationException extends BaseException {
    public ValidationException(String message, String errorCode) {
        super(message, errorCode, 400);
    }
}

Error Propagation Strategy

Errors propagate up the call stack in a layered fashion:

Service layer catches and wraps domain/infrastructure exceptions
Controller layer translates to appropriate HTTP responses
Global exception handler ensures consistent error responses

Retry Strategy with Exponential Backoff and Jitter

For transient infrastructure errors (e.g., database timeouts, external API failures), the system implements a retry mechanism:

public class RetryableOperation {
    private static final int MAX_RETRIES = 3;
    private static final long BASE_DELAY_MS = 1000;
    private static final double JITTER_FACTOR = 0.2;

    public <T> T executeWithRetry(Supplier<T> operation) {
        int attempts = 0;
        while (true) {
            try {
                return operation.get();
            } catch (InfrastructureException e) {
                attempts++;
                if (attempts >= MAX_RETRIES) {
                    throw new InfrastructureException("Operation failed after " + MAX_RETRIES + " attempts", 
                        "OPERATION_FAILED");
                }
                
                long delay = (long) (BASE_DELAY_MS * Math.pow(2, attempts - 1));
                delay += (long) (delay * JITTER_FACTOR * (Math.random() - 0.5) * 2);
                try {
                    Thread.sleep(delay);
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                    throw new InfrastructureException("Retry interrupted", "RETRY_INTERRUPTED");
                }
            }
        }
    }
}

Circuit Breaker Design

For external service dependencies, circuit breaker pattern prevents cascading failures:

public class CircuitBreaker {
    private final int failureThreshold = 5;
    private final long timeoutMs = 60000;
    private final int resetTimeoutMs = 30000;

    private volatile CircuitState state = CircuitState.CLOSED;
    private int failureCount = 0;
    private long lastFailureTime = 0;

    public enum CircuitState {
        CLOSED, OPEN, HALF_OPEN
    }

    public <T> T execute(Supplier<T> operation) {
        switch (state) {
            case CLOSED:
                try {
                    T result = operation.get();
                    onSuccess();
                    return result;
                } catch (Exception e) {
                    onFailure(e);
                    throw e;
                }
            case OPEN:
                if (System.currentTimeMillis() - lastFailureTime > resetTimeoutMs) {
                    state = CircuitState.HALF_OPEN;
                    return operation.get(); // First attempt in half-open
                }
                throw new InfrastructureException("Circuit is OPEN", "CIRCUIT_OPEN");
            case HALF_OPEN:
                try {
                    T result = operation.get();
                    state = CircuitState.CLOSED;
                    return result;
                } catch (Exception e) {
                    state = CircuitState.OPEN;
                    throw e;
                }
        }
        throw new IllegalStateException("Invalid circuit state");
    }

    private void onSuccess() {
        failureCount = 0;
    }

    private void onFailure(Exception e) {
        failureCount++;
        lastFailureTime = System.currentTimeMillis();
        if (failureCount >= failureThreshold) {
            state = CircuitState.OPEN;
        }
    }
}

Fallback Behavior Design

Each external dependency has a fallback implementation:

public class ExternalServiceFallback {
    public String getExternalData(String id) {
        // Return default or cached value
        return "default_value_for_" + id;
    }

    public boolean processPayment(String paymentId) {
        // Log and return false for payment processing
        logger.warn("Payment fallback triggered for ID: {}", paymentId);
        return false;
    }
}

Error Logging Format and Alerting Thresholds

Structured logging includes trace context and error metadata:

public class ErrorLogger {
    private static final Logger logger = LoggerFactory.getLogger(ErrorLogger.class);

    public void logError(String operation, Exception exception, Map<String, Object> context) {
        Map<String, Object> logContext = new HashMap<>(context);
        logContext.put("error_code", extractErrorCode(exception));
        logContext.put("error_type", exception.getClass().getSimpleName());
        logContext.put("timestamp", Instant.now());

        logger.error("Operation {} failed with error: {}", operation, exception.getMessage(), 
            new StructuredLogEvent(logContext));
    }

    private String extractErrorCode(Exception exception) {
        if (exception instanceof BaseException) {
            return ((BaseException) exception).getErrorCode();
        }
        return "UNKNOWN_ERROR";
    }
}

Key metrics thresholds for alerting:

Database timeouts: > 50ms p95 → warn, > 200ms p95 → critical
External service failures: > 5% failure rate → warn, > 10% → critical
Retry attempts: > 3 retries → warn, > 5 retries → critical
Circuit breaker OPEN: > 30s duration → warn, > 2min → critical

11. Security Design

Input Validation and Sanitization Strategy

All inputs are validated using a combination of:

Apache Commons Validator for standard validation (email, URL, phone)
Custom regex patterns for domain-specific constraints
OWASP Java HTML Sanitizer for preventing XSS in text fields

public class InputValidator {
    private static final HtmlSanitizer htmlSanitizer = 
        new HtmlSanitizer.Builder().build();
    
    public static String sanitizeHtml(String input) {
        return htmlSanitizer.sanitize(input);
    }
    
    public static boolean isValidEmail(String email) {
        return EmailValidator.getInstance().isValid(email);
    }
}

SQL Injection Prevention

All database queries use parameterized prepared statements exclusively:

// ❌ BAD - Vulnerable to SQL injection
String query = "SELECT * FROM users WHERE email = '" + email + "'";
Statement stmt = connection.createStatement();
ResultSet rs = stmt.executeQuery(query);

// ✅ GOOD - Parameterized query
String query = "SELECT * FROM users WHERE email = ?";
PreparedStatement stmt = connection.prepareStatement(query);
stmt.setString(1, email);
ResultSet rs = stmt.executeQuery();

Authentication Middleware Design

Custom JWT-based authentication middleware with:

Token expiration handling (1 hour)
Refresh token rotation
Token blacklisting for logout

@Component
public class JwtAuthenticationFilter extends OncePerRequestFilter {
    @Override
    protected void doFilterInternal(HttpServletRequest request, 
                                   HttpServletResponse response,
                                   FilterChain filterChain) throws ServletException, IOException {
        String token = extractToken(request);
        if (token != null && jwtService.validateToken(token)) {
            String username = jwtService.getUsernameFromToken(token);
            UsernamePasswordAuthenticationToken auth = 
                new UsernamePasswordAuthenticationToken(username, null, getAuthorities());
            SecurityContextHolder.getContext().setAuthentication(auth);
        }
        filterChain.doFilter(request, response);
    }
}

Authorization Enforcement

Role-Based Access Control (RBAC) with fine-grained permissions:

@PreAuthorize("hasRole('ADMIN') or @userPermissionService.canAccessUser(principal, userId)")
public void deleteUser(String userId) {
    // Implementation
}

Sensitive Data Encryption

At rest: AES-256-GCM encryption using AWS KMS
In transit: TLS 1.3 with HSTS enabled

@Service
public class DataEncryptionService {
    private final AesGcmEncryptor encryptor = new AesGcmEncryptor();
    
    public String encrypt(String data) {
        return encryptor.encrypt(data, getEncryptionKey());
    }
    
    public String decrypt(String encryptedData) {
        return encryptor.decrypt(encryptedData, getEncryptionKey());
    }
}

PII Handling and Masking Strategy

PII fields (SSN, email, phone) are masked in logs:

public class PiiMasker {
    public static String maskEmail(String email) {
        if (email == null || email.length() < 5) return email;
        String[] parts = email.split("@");
        return parts[0].substring(0, 2) + "****@" + parts[1];
    }
    
    public static String maskPhone(String phone) {
        return phone.replaceAll("(\\d{3})\\d+(\\d{4})", "$1****$2");
    }
}

Secrets Management

Using AWS Secrets Manager with:

Automatic rotation every 90 days
IAM-based access control
Secure key derivation using KMS

Audit Logging Design

All security-relevant events are logged with:

Timestamp and user context
Event type (login, logout, data access)
IP address and session ID

@Component
public class SecurityAuditLogger {
    private static final Logger logger = LoggerFactory.getLogger(SecurityAuditLogger.class);
    
    public void logSecurityEvent(String eventType, String userId, String ipAddress) {
        Map<String, Object> auditEntry = new HashMap<>();
        auditEntry.put("timestamp", Instant.now());
        auditEntry.put("eventType", eventType);
        auditEntry.put("userId", userId);
        auditEntry.put("ipAddress", ipAddress);
        auditEntry.put("sessionId", getSessionId());
        
        logger.info("Security event: {}", auditEntry);
    }
}

12. Testing Strategy

Unit Testing Scope and Coverage Targets

Unit tests will cover 85%+ code coverage for business logic classes, with particular focus on the Chain of Responsibility pipeline components. Each handler in the chain will have dedicated unit tests covering:

Normal execution paths
Error conditions and exception handling
Boundary condition testing (empty inputs, null values)
State transitions for stateful handlers

Integration Testing Strategy

Integration tests will validate the end-to-end pipeline execution with:

Testcontainers for database integration testing (PostgreSQL 13+)
Embedded Redis for cache interaction testing
WireMock for external service stubbing (e.g., payment gateway, notification service)
In-memory Kafka for event-driven component testing

Contract Testing

Contract tests will be implemented using Pact to ensure:

Consumer-provider contract validation between the pipeline service and external dependencies
API boundary testing for HTTP clients that interact with the pipeline
Version compatibility verification for shared data models

Performance Testing Tool and Key Scenarios

Performance testing will be conducted using k6 with the following key scenarios:

High-concurrency pipeline execution: 1000 concurrent requests through the chain with varying handler complexity
Latency-sensitive pipeline: 500 concurrent requests where each handler must complete under 100ms
Memory leak detection: 10,000 sequential pipeline executions to verify no memory accumulation

Test Data Management Strategy

Test data will be managed using:

Factory pattern for generating test data objects (e.g., RequestFactory, ResponseFactory)
Fixture files in JSON/YAML format for complex data structures
Database seeding scripts with predefined test users, configurations, and pipeline rules
Test data isolation using database transaction rollbacks or dedicated test schemas

// Example of factory pattern for testing
public class PipelineRequestFactory {
    public static PipelineRequest createValidRequest() {
        return PipelineRequest.builder()
                .id(UUID.randomUUID().toString())
                .payload("test-payload")
                .metadata(Map.of("source", "test"))
                .build();
    }
    
    public static PipelineRequest createRequestWithInvalidPayload() {
        return PipelineRequest.builder()
                .id(UUID.randomUUID().toString())
                .payload(null)
                .metadata(Map.of("source", "test"))
                .build();
    }
}

13. Observability Design

Structured Logging Schema

All logs follow a structured JSON format with standardized fields to enable efficient querying and analysis:

{
  "timestamp": "2023-10-27T14:30:45.123Z",
  "level": "INFO",
  "service": "payment-service",
  "trace_id": "12345678-1234-1234-1234-123456789012",
  "span_id": "abcdef12-3456-7890-abcd-ef1234567890",
  "user_id": "user_12345",
  "request_id": "req_98765",
  "operation": "processPayment",
  "entity_id": "payment_54321",
  "message": "Payment processing started",
  "duration_ms": 150,
  "error_code": null,
  "error_message": null,
  "stack_trace": null
}

PII masking rules:

User IDs are masked in logs (e.g., user_12345 → user_***)
Credit card numbers are masked (e.g., 1234567890123456 → ****3456)
Email addresses are masked (e.g., john.doe@example.com → j***@example.com)

Distributed Tracing Instrumentation

We use OpenTelemetry for distributed tracing with custom spans and attributes:

Key Operations Instrumented:

Payment Processing Pipeline - Custom span with attributes:
- pipeline.name: "payment-processing"
- pipeline.stage: "validation" | "authorization" | "capture"
- handler.type: "PaymentValidatorHandler" | "PaymentAuthHandler"
- handler.success: true | false
Database Operations - Instrumented with:
- db.system: "postgresql"
- db.name: "payment_db"
- db.statement: "SELECT * FROM payments WHERE id = $1"
External API Calls - Instrumented with:
- http.method: "POST"
- http.url: "https://api.paymentgateway.com/charge"
- http.status_code: 200

Metrics Design

RED Method Metrics (Rate, Errors, Duration):

Metric Name	Type	Labels	Description
`payment_pipeline_requests_total`	Counter	`pipeline`, `stage`, `success`	Total pipeline invocations
`payment_pipeline_errors_total`	Counter	`pipeline`, `stage`, `error_type`	Pipeline error counts
`payment_pipeline_duration_seconds`	Histogram	`pipeline`, `stage`	Pipeline processing duration
`payment_handler_processing_seconds`	Histogram	`handler_type`, `success`	Individual handler processing time
`payment_handler_errors_total`	Counter	`handler_type`, `error_type`	Handler-specific errors

Additional Service Metrics:

active_pipeline_threads (Gauge): Current number of active pipeline workers
pipeline_queue_size (Gauge): Size of pending pipeline jobs queue
failed_pipeline_jobs_total (Counter): Failed pipeline job count

Health Check Endpoint Design

Endpoints:

/health - Liveness check (service is running)
/ready - Readiness check (service can accept traffic)
/live - Startup check (service is fully initialized)
/startup - Detailed startup status

Response Format:

{
  "status": "healthy",
  "timestamp": "2023-10-27T14:30:45.123Z",
  "components": {
    "database": {
      "status": "healthy",
      "latency_ms": 15
    },
    "cache": {
      "status": "healthy",
      "latency_ms": 5
    },
    "external_api": {
      "status": "healthy",
      "latency_ms": 42
    }
  },
  "uptime_seconds": 3600
}

Alerting Thresholds

Critical Alerts (Escalation to On-Call):

payment_pipeline_errors_total > 10 per minute (5 min window)
payment_pipeline_duration_seconds p99 > 5 seconds
database_connection_errors_total > 5 per minute

Warning Alerts:

payment_pipeline_duration_seconds p99 > 2 seconds
pipeline_queue_size > 100 pending jobs
payment_handler_processing_seconds p99 > 1 second

On-Call Escalation Path:

Slack channel #payment-service-alerts
SMS to on-call engineer (after 5 min of no response)
PagerDuty escalation after 10 min of no response

14. Open Questions, Risks & Trade-offs

Decisions requiring product / engineering alignment

Pipeline configuration and dynamic reconfiguration: Should pipeline stages be configurable at runtime via a configuration service, or should they be compiled into the application? Dynamic reconfiguration introduces complexity in managing versioning and rollback strategies.
Stage ordering constraints: Are there strict ordering dependencies between stages that must be enforced at runtime? If so, how do we handle reordering or insertion of new stages without breaking existing workflows?
Stage failure propagation: Should a failed stage in the pipeline halt all subsequent stages, or should it allow downstream stages to continue processing? This affects error handling strategy and user experience.
Monitoring and observability scope: How granular should individual stage monitoring be? Should each stage report metrics independently, or aggregate at the pipeline level?

Known trade-offs and justifications

Flexibility vs. Performance: The Chain of Responsibility pattern provides high flexibility in adding/removing stages but introduces a performance overhead due to multiple method calls and potential context propagation. For high-throughput systems, this trade-off must be carefully evaluated.
Maintainability vs. Complexity: While the pattern promotes loose coupling and single responsibility, it can lead to complex debugging when a failure occurs in an unknown stage. The pipeline structure becomes harder to reason about as stages multiply.
Type Safety vs. Generic Design: Using generic interfaces for stages can reduce type safety and increase runtime errors, but improves reusability across different data types. A strongly-typed approach limits flexibility but provides compile-time guarantees.
Single Responsibility Principle vs. Pipeline State Management: Each stage should handle one concern, but managing shared state across stages requires either passing context objects or maintaining global state, both of which introduce potential race conditions and complexity.

Risks and mitigation strategies

Infinite Loop Risk: If the pipeline design allows stages to call next() in a circular fashion, it can lead to infinite loops. Mitigation includes implementing a maximum stage depth counter and stack trace analysis during pipeline execution.
Resource Exhaustion: Each stage may consume memory or CPU resources, and without proper resource management, the pipeline can cause system instability. Mitigation includes implementing circuit breaker patterns per stage and setting resource limits.
Error Propagation Confusion: When a stage fails, it's unclear whether the error should be handled within that stage or bubbled up to a global error handler. Mitigation involves defining clear error handling contracts and logging strategies for each stage.
Version Compatibility Risk: If stages are updated independently, there's a risk of breaking compatibility with older versions of other stages. Mitigation includes implementing versioned interfaces and strict backward compatibility policies.

Technical debt acknowledged

Hardcoded Stage Dependencies: The current design assumes all stages are known at compile time, which makes it difficult to support plugin architectures or runtime stage addition without refactoring.
Limited Stage Metadata Support: There's no built-in support for metadata like stage priority, retry logic, or timeout configuration within the pipeline itself, requiring external configuration management systems.
No Built-in Retry Mechanism: The pattern does not inherently support retry logic or circuit breaker patterns, which must be manually implemented in each stage or handled at a higher level.
Context Propagation Complexity: As pipelines grow, managing context propagation across stages becomes increasingly complex and error-prone without proper abstraction or framework support.
Testing Challenges: Testing individual stages in isolation becomes harder as the pipeline grows, requiring extensive mocking and integration testing to ensure proper stage chaining behavior.