Advanced Circuit Breaker Patterns
After reviewing the basic try/catch and fallback mechanisms, use these variants as adaptation sketches. Their runtime behavior has not been assessed by a current machine gate.
Pattern 1: Circuit Breaker with Caching
For read-heavy operations, caching is the most effective way to reduce load on downstream systems and improve resilience. If the primary data source is unavailable, the pipeline can fall back to serving slightly stale data from the cache.
Use Case: Enriching events with user metadata that changes infrequently.
Configuration
This pipeline tries to fetch data from a cache first. On a cache miss it attempts the database call, then writes the result to the configured cache after a successful call.
config:
# Define a cache resource
cache_resources:
- label: user_profile_cache
memory:
default_ttl: 5m # Cache entries for 5 minutes
pipeline:
processors:
# 1. Check the cache first
- branch:
processors:
- branch:
processors:
- cache:
resource: user_profile_cache
operator: exists
key: user_${!this.user_id}
result_map: root.profile_exists = content().string().bool()
- switch:
- check: this.profile_exists
processors:
- try:
- cache:
resource: user_profile_cache
operator: get
key: user_${!this.user_id}
- mapping: |
root.profile = this
root.found = true
- catch:
- mapping: |
root = {"found": false}
- processors:
- mapping: |
root = {"found": false}
result_map: |
root.profile_cached = this.found
root.user_profile = if this.found { this.profile } else { deleted() }
# 2. Act based on cache hit or miss
- switch:
# CACHE HIT: Use the cached data
- check: this.profile_cached
processors:
- mapping: |
root = this
root.data_source = "cache"
# CACHE MISS: Query the database
- processors:
- try:
- branch:
processors:
- sql_select:
driver: postgres
dsn: ${DB_CONNECTION_STRING}
table: users
columns: [user_name, user_tier]
where: user_id = $1
args_mapping: root = [ this.user_id ]
result_map: |
root.user_name = this.index(0).user_name
root.user_tier = this.index(0).user_tier
- mapping: |
root = this
root.user_profile = { "name": this.user_name, "tier": this.user_tier }
root.data_source = "database"
# On success, write the result to the cache
- cache:
resource: user_profile_cache
operator: set
key: user_${!this.user_id}
value: ${!this.user_profile.format_json()}
- catch:
# If the DB fails, mark it and continue without enrichment
- mapping: |
root = this
root.data_source = "none"
root.db_error = true
Pattern 2: Read-Write Separation
Write operations (e.g., INSERT, UPDATE) are often more critical and have different failure modes than read operations (SELECT). It's a good practice to use separate, stricter circuit breaker policies for writes.
Use Case: A pipeline that both reads user preferences and writes new user events to a database.
Configuration
This example uses a switch to route events based on an operation field. Writes get a very conservative circuit breaker, while reads are more tolerant.
pipeline:
processors:
- switch:
# READ operations have a tolerant circuit breaker
- check: this.operation == "read"
processors:
- try:
- sql_select:
driver: postgres
dsn: ${DB_CONNECTION_STRING}
table: users
columns: [user_name, user_tier]
where: user_id = $1
args_mapping: root = [this.user_id]
- catch:
- mapping: |
root = this
root.read_failed = true
# WRITE operations have a strict circuit breaker
- check: this.operation == "write"
processors:
- try:
- sql_insert:
driver: postgres
dsn: ${DB_CONNECTION_STRING}
table: events
columns: [event_id, payload]
args_mapping: root = [this.event_id, this.format_json()]
- catch:
- mapping: |
root = this
root.write_failed = true
output:
switch:
cases:
# Failed writes are sent to a retry queue
- check: this.write_failed == true
output:
kafka:
addresses: ['${KAFKA_BROKERS}']
topic: 'database_writes_retry'
# All other events go to a standard output
- output:
stdout: {}
Pattern 3: Adaptive Circuit Breaker
An adaptive circuit breaker can dynamically change its behavior based on the health of a downstream system. This is an advanced technique that uses a lightweight health check to decide which circuit breaker policy to apply.
Use Case: Before calling a critical, expensive API, first ping a lightweight /health endpoint on that same API.
Configuration
pipeline:
processors:
# 1. Perform a quick health check
- try:
- branch:
processors:
- http:
url: ${MOCK_API_URL}/health
verb: GET
timeout: 1s
result_map: root.api_is_healthy = true
- catch:
- mapping: |
root = this
root.api_is_healthy = false
# 2. Choose a breaker policy based on health
- switch:
# If healthy, use normal settings
- check: this.api_is_healthy == true
processors:
- http:
url: ${MOCK_API_URL}/metadata/${!this.sensor_id}
timeout: 10s
retries: 5
# If unhealthy, use aggressive settings to fail fast
- processors:
- http:
url: ${MOCK_API_URL}/metadata/${!this.sensor_id}
timeout: 2s
retries: 1
These advanced patterns add explicit fallback and recovery controls. Validate their resilience and performance in the target environment.
