Hystrix, Resilience4j

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53

import com.netflix.hystrix.HystrixCommand;
import com.netflix.hystrix.HystrixCommandGroupKey;
import com.netflix.hystrix.HystrixCommandProperties;
import com.netflix.hystrix.HystrixThreadPoolProperties;
 
public class GetProductCommand extends HystrixCommand<Product> {
    
    private final ProductService productService;
    private final String productId;
    
    public GetProductCommand(ProductService productService, String productId) {
        super(Setter
            .withGroupKey(HystrixCommandGroupKey.Factory.asKey("ProductService"))
            .andCommandPropertiesDefaults(
                HystrixCommandProperties.Setter()
                    // Circuit breaker settings
                    .withCircuitBreakerEnabled(true)
                    .withCircuitBreakerRequestVolumeThreshold(20)  // Min requests before evaluating
                    .withCircuitBreakerErrorThresholdPercentage(50) // Trip at 50% failure
                    .withCircuitBreakerSleepWindowInMilliseconds(5000) // Recovery timeout
                    
                    // Timeout settings
                    .withExecutionTimeoutInMilliseconds(1000)
                    
                    // Metrics settings
                    .withMetricsRollingStatisticalWindowInMilliseconds(10000) // 10s window
            )
            .andThreadPoolPropertiesDefaults(
                HystrixThreadPoolProperties.Setter()
                    .withCoreSize(10)  // Thread pool size
                    .withMaxQueueSize(100)
                    .withQueueSizeRejectionThreshold(100)
            )
        );
        this.productService = productService;
        this.productId = productId;
    }
    
    @Override
    protected Product run() throws Exception {
        // The actual call to the external service
        return productService.getProduct(productId);
    }
    
    @Override
    protected Product getFallback() {
        // Fallback when circuit is open or call fails
        return Product.createCachedOrDefault(productId);
    }
}
 
// Usage
Product product = new GetProductCommand(productService, "prod-123").execute();

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

import io.github.resilience4j.circuitbreaker.CircuitBreaker;
import io.github.resilience4j.circuitbreaker.CircuitBreakerConfig;
import io.github.resilience4j.circuitbreaker.CircuitBreakerRegistry;
 
// 1. Create configuration
CircuitBreakerConfig config = CircuitBreakerConfig.custom()
    .failureRateThreshold(50)
    .minimumNumberOfCalls(10)
    .waitDurationInOpenState(Duration.ofSeconds(30))
    .permittedNumberOfCallsInHalfOpenState(5)
    .slidingWindowType(SlidingWindowType.COUNT_BASED)
    .slidingWindowSize(100)
    .slowCallRateThreshold(80)
    .slowCallDurationThreshold(Duration.ofSeconds(2))
    .build();
 
// 2. Create registry and circuit breaker
CircuitBreakerRegistry registry = CircuitBreakerRegistry.of(config);
CircuitBreaker circuitBreaker = registry.circuitBreaker("productService");
 
// 3. Decorate your function
Supplier<Product> decoratedSupplier = CircuitBreaker
    .decorateSupplier(circuitBreaker, () -> productService.getProduct(productId));
 
// 4. Execute with fallback
Product product = Try.ofSupplier(decoratedSupplier)
    .recover(throwable -> Product.createDefault(productId))
    .get();
 
// Alternative: Functional composition
Product product = circuitBreaker.executeSupplier(
    () -> productService.getProduct(productId)
);
 
// With fallback using Decorators
Supplier<Product> supplier = Decorators.ofSupplier(() -> productService.getProduct(productId))
    .withCircuitBreaker(circuitBreaker)
    .withFallback(Arrays.asList(CallNotPermittedException.class, Exception.class),
                  e -> Product.createDefault(productId))
    .decorate();

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

// Create individual components
CircuitBreaker circuitBreaker = CircuitBreaker.ofDefaults("product");
RateLimiter rateLimiter = RateLimiter.ofDefaults("product");
Retry retry = Retry.ofDefaults("product");
Bulkhead bulkhead = Bulkhead.ofDefaults("product");
TimeLimiter timeLimiter = TimeLimiter.ofDefaults("product");
 
// Compose them using Decorators utility
Supplier<CompletionStage<Product>> decoratedSupplier = Decorators
    .ofSupplier(() -> productService.getProduct(productId))
    .withRetry(retry)                  // Outermost: retry failed attempts
    .withCircuitBreaker(circuitBreaker) // Next: check circuit state
    .withBulkhead(bulkhead)            // Next: acquire bulkhead permit
    .withRateLimiter(rateLimiter)      // Next: check rate limit
    .withFallback(throwable -> Product.createDefault(productId))
    .decorate();
 
/*
 * Order of decorators matters!
 * 
 * The call flows through decorators outside-in:
 *   Retry → CircuitBreaker → Bulkhead → RateLimiter → Actual Call
 * 
 * Results/exceptions flow back inside-out:
 *   Actual Call → RateLimiter → Bulkhead → CircuitBreaker → Retry
 * 
 * Typical recommended order:
 *   1. Retry (outside) - retries should wrap everything
 *   2. CircuitBreaker - should NOT retry when open
 *   3. RateLimiter/Bulkhead - protect resources
 *   4. TimeLimiter - bound execution time
 *   5. Actual call (inside)
 */

1
2
3
4
5
6
7
8
9
10
11
12
13
14

dependencies {
    // Core Spring Boot
    implementation 'org.springframework.boot:spring-boot-starter-web'
    implementation 'org.springframework.boot:spring-boot-starter-aop'
    
    // Resilience4j Spring Boot Starter
    implementation 'io.github.resilience4j:resilience4j-spring-boot2:2.1.0'
    
    // Micrometer for metrics (recommended)
    implementation 'io.micrometer:micrometer-registry-prometheus'
    
    // Actuator for health checks
    implementation 'org.springframework.boot:spring-boot-starter-actuator'
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

resilience4j:
  circuitbreaker:
    configs:
      default:
        registerHealthIndicator: true
        slidingWindowSize: 100
        minimumNumberOfCalls: 10
        permittedNumberOfCallsInHalfOpenState: 5
        automaticTransitionFromOpenToHalfOpenEnabled: true
        waitDurationInOpenState: 30s
        failureRateThreshold: 50
        slowCallRateThreshold: 80
        slowCallDurationThreshold: 2s
        eventConsumerBufferSize: 10
        recordExceptions:
          - org.springframework.web.client.HttpServerErrorException
          - java.io.IOException
          - java.util.concurrent.TimeoutException
        ignoreExceptions:
          - com.example.BusinessException
      
      # Custom configuration for payment service (more conservative)
      payment-critical:
        failureRateThreshold: 30
        slowCallRateThreshold: 60
        waitDurationInOpenState: 60s
        minimumNumberOfCalls: 5
    
    instances:
      # Apply 'default' config to product service
      productService:
        baseConfig: default
      
      # Apply 'payment-critical' config to payment service
      paymentService:
        baseConfig: payment-critical
        # Override specific settings
        waitDurationInOpenState: 45s
  
  # Rate limiter configuration
  ratelimiter:
    instances:
      productService:
        limitForPeriod: 100
        limitRefreshPeriod: 1s
        timeoutDuration: 500ms
  
  # Retry configuration
  retry:
    instances:
      productService:
        maxAttempts: 3
        waitDuration: 500ms
        retryExceptions:
          - java.io.IOException
          - java.util.concurrent.TimeoutException

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

import io.github.resilience4j.circuitbreaker.annotation.CircuitBreaker;
import io.github.resilience4j.retry.annotation.Retry;
import io.github.resilience4j.ratelimiter.annotation.RateLimiter;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestTemplate;
 
@Service
public class ProductServiceClient {
    
    private final RestTemplate restTemplate;
    
    public ProductServiceClient(RestTemplate restTemplate) {
        this.restTemplate = restTemplate;
    }
    
    // Apply circuit breaker, retry, and rate limiter
    // Order: Retry wraps CircuitBreaker wraps RateLimiter
    @CircuitBreaker(name = "productService", fallbackMethod = "getProductFallback")
    @Retry(name = "productService")
    @RateLimiter(name = "productService")
    public Product getProduct(String productId) {
        return restTemplate.getForObject(
            "http://product-service/products/{id}",
            Product.class,
            productId
        );
    }
    
    // Fallback method - must have same signature + Throwable
    private Product getProductFallback(String productId, Throwable throwable) {
        log.warn("Fallback triggered for product {}: {}", 
                 productId, throwable.getMessage());
        
        if (throwable instanceof CallNotPermittedException) {
            // Circuit is open - use cached data
            return productCache.get(productId)
                .orElse(Product.unavailable(productId));
        }
        
        // Other failures - return default
        return Product.createDefault(productId);
    }
    
    // You can have different fallback methods for different scenarios
    @CircuitBreaker(name = "productService", 
                   fallbackMethod = "getProductsFallback")
    public List<Product> getProducts(List<String> productIds) {
        // ...
    }
    
    private List<Product> getProductsFallback(List<String> productIds, 
                                              Throwable throwable) {
        return productIds.stream()
            .map(Product::unavailable)
            .collect(Collectors.toList());
    }
}

# Health endpoint includes circuit breaker status
GET /actuator/health
 
{
  "status": "UP",
  "components": {
    "circuitBreakers": {
      "status": "UP",
      "details": {
        "productService": {
          "status": "UP",
          "details": {
            "failureRate": "0.0%",
            "slowCallRate": "0.0%",
            "slowCallRateThreshold": "80.0%",
            "failureRateThreshold": "50.0%",
            "bufferedCalls": 45,
            "failedCalls": 0,
            "slowCalls": 2,
            "slowFailedCalls": 0,
            "notPermittedCalls": 0,
            "state": "CLOSED"
          }
        },
        "paymentService": {
          "status": "CIRCUIT_OPEN",
          "details": {
            "failureRate": "67.5%",
            "state": "OPEN"
          }
        }
      }
    }
  }
}
 
# Dedicated circuit breaker events endpoint
GET /actuator/circuitbreakerevents
 
# Filter events by circuit breaker name
GET /actuator/circuitbreakerevents/productService
 
# Filter by event type
GET /actuator/circuitbreakerevents/productService/error

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62

import CircuitBreaker from 'opossum';
 
// Define the function to protect
async function fetchUser(userId: string): Promise<User> {
    const response = await fetch(`https://api.example.com/users/${userId}`);
    if (!response.ok) throw new Error(`HTTP ${response.status}`);
    return response.json();
}
 
// Create circuit breaker
const breaker = new CircuitBreaker(fetchUser, {
    // Trip conditions
    errorThresholdPercentage: 50, // Open at 50% error rate
    volumeThreshold: 10,          // Minimum calls to evaluate
    
    // Timing
    timeout: 5000,                // Call timeout
    resetTimeout: 30000,          // Recovery timeout
    
    // Rolling window
    rollingCountTimeout: 10000,   // 10-second rolling window
    rollingCountBuckets: 10,      // 10 buckets of 1 second each
});
 
// Event handlers for monitoring
breaker.on('success', (result) => {
    metrics.increment('circuitbreaker.success');
});
 
breaker.on('failure', (error) => {
    metrics.increment('circuitbreaker.failure');
    console.error('Call failed:', error);
});
 
breaker.on('open', () => {
    metrics.increment('circuitbreaker.opened');
    console.warn('Circuit opened');
});
 
breaker.on('halfOpen', () => {
    console.log('Testing recovery...');
});
 
breaker.on('close', () => {
    metrics.increment('circuitbreaker.closed');
    console.log('Circuit closed - recovered');
});
 
// Fallback
breaker.fallback((userId) => ({
    id: userId,
    name: 'Unknown User',
    status: 'unavailable'
}));
 
// Usage
try {
    const user = await breaker.fire('user-123');
    console.log(user);
} catch (error) {
    console.error('Failed with fallback');
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

package main
 
import (
    "fmt"
    "time"
    
    "github.com/sony/gobreaker"
)
 
func main() {
    settings := gobreaker.Settings{
        Name:        "ProductService",
        MaxRequests: 5,                    // Max requests in half-open
        Interval:    30 * time.Second,     // Failure count reset interval
        Timeout:     60 * time.Second,     // Recovery timeout
        
        // Custom trip condition
        ReadyToTrip: func(counts gobreaker.Counts) bool {
            failureRatio := float64(counts.TotalFailures) / float64(counts.Requests)
            return counts.Requests >= 10 && failureRatio >= 0.5
        },
        
        // Event handlers
        OnStateChange: func(name string, from, to gobreaker.State) {
            fmt.Printf("Circuit %s: %s -> %s
", name, from, to)
        },
    }
    
    cb := gobreaker.NewCircuitBreaker(settings)
    
    // Execute protected function
    result, err := cb.Execute(func() (interface{}, error) {
        return fetchProduct("prod-123")
    })
    
    if err != nil {
        if err == gobreaker.ErrOpenState {
            // Circuit is open
            fmt.Println("Circuit open, using fallback")
            result = getProductFallback("prod-123")
        } else {
            // Other error
            fmt.Println("Error:", err)
        }
    }
    
    fmt.Println("Result:", result)
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

from pybreaker import CircuitBreaker, CircuitBreakerError
import requests
 
# Create circuit breaker
product_breaker = CircuitBreaker(
    fail_max=5,           # Failures to trip
    reset_timeout=30,     # Recovery timeout in seconds
    exclude=[ValueError], # Exceptions to ignore
    listeners=[           # Event listeners
        LoggingListener(),
        MetricsListener()
    ]
)
 
@product_breaker
def get_product(product_id: str) -> dict:
    """Protected function - decorated with circuit breaker."""
    response = requests.get(
        f"http://product-service/products/{product_id}",
        timeout=5
    )
    response.raise_for_status()
    return response.json()
 
def get_product_with_fallback(product_id: str) -> dict:
    """Wrapper with fallback handling."""
    try:
        return get_product(product_id)
    except CircuitBreakerError:
        # Circuit is open
        return {"id": product_id, "status": "unavailable", "cached": True}
    except requests.RequestException as e:
        # Call failed (circuit might trip if threshold reached)
        return {"id": product_id, "status": "error", "error": str(e)}
 
# Alternative: circuitbreaker library (simpler)
from circuitbreaker import circuit
 
@circuit(failure_threshold=5, recovery_timeout=30)
def fetch_user(user_id: str) -> dict:
    response = requests.get(f"http://user-service/users/{user_id}")
    response.raise_for_status()
    return response.json()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50

using Polly;
using Polly.CircuitBreaker;
 
// Define the circuit breaker policy
var circuitBreakerPolicy = Policy
    .Handle<HttpRequestException>()
    .OrResult<HttpResponseMessage>(r => !r.IsSuccessStatusCode)
    .AdvancedCircuitBreakerAsync(
        failureThreshold: 0.5,              // 50% failure rate to trip
        samplingDuration: TimeSpan.FromSeconds(30), // Sampling window
        minimumThroughput: 10,              // Minimum calls to evaluate
        durationOfBreak: TimeSpan.FromSeconds(30),  // Recovery timeout
        onBreak: (exception, duration) =>
        {
            _logger.LogWarning($"Circuit opened for {duration.TotalSeconds}s");
        },
        onReset: () =>
        {
            _logger.LogInformation("Circuit closed");
        },
        onHalfOpen: () =>
        {
            _logger.LogInformation("Circuit half-open");
        }
    );
 
// Combine with retry and fallback
var resilientPolicy = Policy.WrapAsync(
    Policy<Product>
        .Handle<Exception>()
        .FallbackAsync(
            fallbackValue: Product.CreateDefault(),
            onFallbackAsync: async (exception, context) =>
            {
                _logger.LogWarning($"Fallback triggered: {exception.Exception.Message}");
            }
        ),
    Policy
        .Handle<HttpRequestException>()
        .WaitAndRetryAsync(3, retry => TimeSpan.FromSeconds(Math.Pow(2, retry))),
    circuitBreakerPolicy
);
 
// Execute
var product = await resilientPolicy.ExecuteAsync(async () =>
{
    var response = await _httpClient.GetAsync($"/products/{productId}");
    response.EnsureSuccessStatusCode();
    return await response.Content.ReadAsAsync<Product>();
});

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

apiVersion: networking.istio.io/v1beta1
kind: DestinationRule
metadata:
  name: product-service
spec:
  host: product-service.default.svc.cluster.local
  trafficPolicy:
    connectionPool:
      tcp:
        maxConnections: 100          # Connection pool limit
      http:
        h2UpgradePolicy: UPGRADE
        http1MaxPendingRequests: 100  # Queue limit
        http2MaxRequests: 1000        # Concurrent requests limit
        maxRequestsPerConnection: 100
        maxRetries: 3
    
    outlierDetection:
      # These settings implement circuit-breaker-like behavior
      consecutive5xxErrors: 5         # Eject after 5 consecutive 5xx
      interval: 30s                   # Ejection evaluation interval
      baseEjectionTime: 30s           # Minimum ejection duration
      maxEjectionPercent: 50          # Max % of hosts to eject
      minHealthPercent: 30            # Min healthy hosts before panic
      
      # Can also use consecutive gateway errors
      consecutiveGatewayErrors: 5
      
      # Or track all errors
      splitExternalLocalOriginErrors: true
      consecutiveLocalOriginFailures: 5

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

// BEFORE: Hystrix Command
public class GetProductCommand extends HystrixCommand<Product> {
    
    private final ProductService productService;
    private final String productId;
    
    public GetProductCommand(ProductService productService, String productId) {
        super(Setter
            .withGroupKey(HystrixCommandGroupKey.Factory.asKey("ProductService"))
            .andCommandPropertiesDefaults(
                HystrixCommandProperties.Setter()
                    .withCircuitBreakerRequestVolumeThreshold(20)
                    .withCircuitBreakerErrorThresholdPercentage(50)
                    .withCircuitBreakerSleepWindowInMilliseconds(5000)
                    .withExecutionTimeoutInMilliseconds(3000)
            )
            .andThreadPoolPropertiesDefaults(
                HystrixThreadPoolProperties.Setter()
                    .withCoreSize(10)
            )
        );
        this.productService = productService;
        this.productId = productId;
    }
    
    @Override
    protected Product run() throws Exception {
        return productService.getProduct(productId);
    }
    
    @Override
    protected Product getFallback() {
        return Product.createDefault(productId);
    }
}
 
// Usage
Product product = new GetProductCommand(productService, productId).execute();