3.4 Backpressure Control

Overview

Backpressure control is a critical mechanism for safely managing requests that exceed system capacity. It prevents system crashes during high traffic scenarios while maintaining performance and reliability.

Quick Decision

Situation	Backpressure	Watch Out
Downstream capacity is limited	Rate limit / queue limit	Limit response should be clear to clients
Queue keeps growing	Reject or shed load	Infinite queues kill systems
Traffic has short bursts	Token bucket	Burst capacity must be measured
Non-critical workload	Drop/defer	Priority policy is needed

Production Checklist

• Problem: Which resource is saturated: thread, DB, queue, CPU, or external service?
• Solution: Are limit, queue size, timeout, retry, and 429/503 behavior clear?
• Trade-off: Rejecting requests looks bad; crashing the system is worse.
• Failure mode: Retry storms, queue overflow, thread starvation, and unfair tenant usage should be handled.
• Measurement: Track queue depth, reject rate, saturation, retry count, and p99 latency.
• Security/cost: Limits should be tenant-aware; unlimited capacity increases hide cost.

Rate Limiting

Token Bucket Algorithm

@Component
public class TokenBucketRateLimiter {
    
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    
    public boolean isAllowed(String key, int capacity, int refillRate) {
        String script = """
            local key = KEYS[1]
            local capacity = tonumber(ARGV[1])
            local tokens = tonumber(ARGV[2])
            local interval = tonumber(ARGV[3])
            
            local bucket = redis.call('hmget', key, 'tokens', 'last_refill')
            local tokens_count = tonumber(bucket[1]) or capacity
            local last_refill = tonumber(bucket[2]) or redis.call('time')[1]
            
            local now = redis.call('time')[1]
            local elapsed = now - last_refill
            local tokens_to_add = math.floor(elapsed * tokens / interval)
            
            tokens_count = math.min(capacity, tokens_count + tokens_to_add)
            
            if tokens_count >= 1 then
                tokens_count = tokens_count - 1
                redis.call('hmset', key, 'tokens', tokens_count, 'last_refill', now)
                redis.call('expire', key, interval * 2)
                return 1
            else
                redis.call('hmset', key, 'tokens', tokens_count, 'last_refill', now)
                redis.call('expire', key, interval * 2)
                return 0
            end
        """;
        
        List<String> keys = Arrays.asList(key);
        List<Object> args = Arrays.asList(capacity, refillRate, 60);
        
        Long result = (Long) redisTemplate.execute(
            (RedisCallback<Long>) connection -> 
                connection.eval(script.getBytes(), 
                    ReturnType.INTEGER, 
                    keys.size(), 
                    keys.toArray(new String[0]),
                    args.toArray(new String[0])
                )
        );
        
        return result != null && result == 1;
    }
}

API Gateway Rate Limiting

@RestController
@RequestMapping("/api")
public class RateLimitedController {
    
    @Autowired
    private TokenBucketRateLimiter rateLimiter;
    
    @GetMapping("/data")
    public ResponseEntity<?> getData(HttpServletRequest request) {
        String clientId = getClientId(request);
        String key = "rate_limit:" + clientId;
        
        // 100 requests per minute per client
        if (!rateLimiter.isAllowed(key, 100, 100)) {
            HttpHeaders headers = new HttpHeaders();
            headers.add("X-RateLimit-Limit", "100");
            headers.add("X-RateLimit-Remaining", "0");
            headers.add("Retry-After", "60");
            
            return ResponseEntity.status(HttpStatus.TOO_MANY_REQUESTS)
                .headers(headers)
                .body(Map.of("error", "Rate limit exceeded"));
        }
        
        return ResponseEntity.ok(processData());
    }
    
    private String getClientId(HttpServletRequest request) {
        String apiKey = request.getHeader("X-API-Key");
        if (apiKey != null) {
            return apiKey;
        }
        return request.getRemoteAddr();
    }
}

Queue Management

Message Queue Backpressure

@Configuration
@EnableRabbitMQ
public class QueueBackpressureConfig {
    
    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate template = new RabbitTemplate(connectionFactory);
        
        // Publisher confirms callback
        template.setConfirmCallback((correlationData, ack, cause) -> {
            if (!ack) {
                log.error("Message rejected: {}", cause);
                // Retry logic or dead letter queue
            }
        });
        
        // Return callback for unroutable messages
        template.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
            log.error("Message returned: {} - {}", replyCode, replyText);
        });
        
        return template;
    }
    
    @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(
            ConnectionFactory connectionFactory) {
        
        SimpleRabbitListenerContainerFactory factory = 
            new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        
        // Backpressure control
        factory.setPrefetchCount(10); // Max 10 messages per consumer
        factory.setConcurrentConsumers(2);
        factory.setMaxConcurrentConsumers(10);
        
        // Error handling
        factory.setErrorHandler(new ConditionalRejectingErrorHandler(
            new ConditionalRejectingErrorHandler.DefaultExceptionStrategy())
        );
        
        return factory;
    }
}

Async Processing with Backpressure

@Service
public class BackpressureAwareProcessor {
    
    private final Semaphore semaphore;
    private final ExecutorService executorService;
    
    public BackpressureAwareProcessor() {
        this.semaphore = new Semaphore(100); // Max 100 concurrent operations
        this.executorService = Executors.newFixedThreadPool(20);
    }
    
    @Async
    public CompletableFuture<String> processWithBackpressure(String data) {
        return CompletableFuture.supplyAsync(() -> {
            try {
                // Semaphore for concurrency control
                if (!semaphore.tryAcquire(1, TimeUnit.SECONDS)) {
                    throw new BackpressureException("System overloaded");
                }
                
                try {
                    return processData(data);
                } finally {
                    semaphore.release();
                }
                
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new RuntimeException(e);
            }
        }, executorService);
    }
    
    private String processData(String data) {
        // Simulate processing time
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        return "Processed: " + data;
    }
}

@ResponseStatus(HttpStatus.SERVICE_UNAVAILABLE)
public class BackpressureException extends RuntimeException {
    public BackpressureException(String message) {
        super(message);
    }
}

Circuit Breaking

Resilience4j Integration

@Configuration
public class CircuitBreakerConfig {
    
    @Bean
    public CircuitBreaker databaseCircuitBreaker() {
        return CircuitBreaker.ofDefaults("database");
    }
    
    @Bean
    public CircuitBreakerRegistry circuitBreakerRegistry() {
        CircuitBreakerConfig config = CircuitBreakerConfig.custom()
            .failureRateThreshold(50) // 50% failure rate
            .waitDurationInOpenState(Duration.ofSeconds(30))
            .slidingWindowSize(10)
            .minimumNumberOfCalls(5)
            .slowCallRateThreshold(50)
            .slowCallDurationThreshold(Duration.ofSeconds(2))
            .build();
            
        return CircuitBreakerRegistry.of(config);
    }
}

@Service
public class ExternalServiceClient {
    
    private final CircuitBreaker circuitBreaker;
    private final TimeLimiter timeLimiter;
    private final Retry retry;
    
    public ExternalServiceClient(CircuitBreakerRegistry registry) {
        this.circuitBreaker = registry.circuitBreaker("external-service");
        this.timeLimiter = TimeLimiter.ofDefaults();
        this.retry = Retry.ofDefaults("external-service");
    }
    
    public CompletableFuture<String> callExternalService(String data) {
        Supplier<CompletableFuture<String>> decoratedSupplier = 
            Decorators.ofSupplier(() -> CompletableFuture.supplyAsync(() -> {
                // External service call simulation
                if (Math.random() > 0.7) {
                    throw new RuntimeException("Service unavailable");
                }
                return "Response for: " + data;
            }))
            .withCircuitBreaker(circuitBreaker)
            .withTimeLimiter(timeLimiter)
            .withRetry(retry)
            .decorate();
            
        return decoratedSupplier.get();
    }
}

Connection Pooling

Database Connection Management

@Configuration
public class DatabasePoolConfig {
    
    @Bean
    @Primary
    public DataSource dataSource() {
        HikariConfig config = new HikariConfig();
        config.setJdbcUrl("jdbc:postgresql://localhost:5432/mydb");
        config.setUsername("user");
        config.setPassword("password");
        
        // Connection pool settings
        config.setMaximumPoolSize(20);
        config.setMinimumIdle(5);
        config.setConnectionTimeout(30000); // 30 seconds
        config.setIdleTimeout(600000); // 10 minutes
        config.setMaxLifetime(1800000); // 30 minutes
        
        // Backpressure settings
        config.setLeakDetectionThreshold(60000); // 1 minute
        
        return new HikariDataSource(config);
    }
    
    @Bean
    public JdbcTemplate jdbcTemplate(DataSource dataSource) {
        JdbcTemplate template = new JdbcTemplate(dataSource);
        template.setQueryTimeout(30); // 30 seconds query timeout
        return template;
    }
}

HTTP Client Pool Management

@Configuration
public class HttpClientConfig {
    
    @Bean
    public RestTemplate restTemplate() {
        HttpComponentsClientHttpRequestFactory factory = 
            new HttpComponentsClientHttpRequestFactory();
            
        // Connection pool configuration
        factory.setConnectTimeout(5000); // 5 seconds
        factory.setReadTimeout(10000); // 10 seconds
        
        CloseableHttpClient httpClient = HttpClients.custom()
            .setMaxConnTotal(100) // Total max connections
            .setMaxConnPerRoute(20) // Max connections per route
            .setConnectionTimeToLive(30, TimeUnit.SECONDS)
            .build();
            
        factory.setHttpClient(httpClient);
        
        RestTemplate restTemplate = new RestTemplate(factory);
        
        // Error handling for backpressure
        restTemplate.setErrorHandler(new ResponseErrorHandler() {
            @Override
            public boolean hasError(ClientHttpResponse response) throws IOException {
                return response.getStatusCode().series() == 
                    HttpStatus.Series.CLIENT_ERROR ||
                    response.getStatusCode().series() == 
                    HttpStatus.Series.SERVER_ERROR;
            }
            
            @Override
            public void handleError(ClientHttpResponse response) throws IOException {
                if (response.getStatusCode() == HttpStatus.TOO_MANY_REQUESTS) {
                    throw new BackpressureException("Rate limit exceeded");
                }
                throw new RestClientException("HTTP error: " + 
                    response.getStatusCode());
            }
        });
        
        return restTemplate;
    }
}

Monitoring and Alerting

Backpressure Metrics

@Component
public class BackpressureMetrics {
    
    private final MeterRegistry meterRegistry;
    private final Counter rejectedRequests;
    private final Gauge queueSize;
    private final Timer processingTime;
    
    public BackpressureMetrics(MeterRegistry meterRegistry) {
        this.meterRegistry = meterRegistry;
        this.rejectedRequests = Counter.builder("requests.rejected")
            .description("Number of rejected requests due to backpressure")
            .register(meterRegistry);
            
        this.queueSize = Gauge.builder("queue.size")
            .description("Current queue size")
            .register(meterRegistry, this, BackpressureMetrics::getCurrentQueueSize);
            
        this.processingTime = Timer.builder("processing.time")
            .description("Request processing time")
            .register(meterRegistry);
    }
    
    public void recordRejection(String reason) {
        rejectedRequests.increment(Tags.of("reason", reason));
    }
    
    public Timer.Sample startTimer() {
        return Timer.start(meterRegistry);
    }
    
    private double getCurrentQueueSize() {
        // Implementation to get current queue size
        return 0.0;
    }
}

Health Check Integration

@Component
public class BackpressureHealthIndicator implements HealthIndicator {
    
    private final BackpressureAwareProcessor processor;
    private final TokenBucketRateLimiter rateLimiter;
    
    public BackpressureHealthIndicator(
            BackpressureAwareProcessor processor,
            TokenBucketRateLimiter rateLimiter) {
        this.processor = processor;
        this.rateLimiter = rateLimiter;
    }
    
    @Override
    public Health health() {
        Health.Builder builder = new Health.Builder();
        
        try {
            // Check system capacity
            boolean hasCapacity = checkSystemCapacity();
            
            if (hasCapacity) {
                builder.up()
                    .withDetail("status", "System has available capacity")
                    .withDetail("available_permits", getAvailablePermits());
            } else {
                builder.down()
                    .withDetail("status", "System at capacity")
                    .withDetail("available_permits", 0);
            }
            
        } catch (Exception e) {
            builder.down()
                .withDetail("error", e.getMessage());
        }
        
        return builder.build();
    }
    
    private boolean checkSystemCapacity() {
        // Implementation to check system capacity
        return true;
    }
    
    private int getAvailablePermits() {
        // Implementation to get available permits
        return 50;
    }
}

This backpressure control implementation ensures systems operate safely under overload conditions and prevents performance degradation.