backend / message queues / celery / 02_how_to_make_retry_in_celery.md

How to Make Retry in Celery

4 interview angles 6 min read source

How to Make Retry in Celery

Overview

Celery provides robust retry mechanisms to handle task failures gracefully. Retries are essential for dealing with transient errors, network issues, or temporary resource unavailability.

Basic Retry Implementation

Simple Retry with Decorator

from celery import Celery

app = Celery('tasks', broker='redis://localhost:6379/0')

@app.task(bind=True, max_retries=3)
def my_task(self, x, y):
    try:
        result = x / y
        return result
    except ZeroDivisionError as exc:
        # Retry the task
        raise self.retry(exc=exc, countdown=60)

Key parameters:

  • bind=True: Makes the task instance available as self
  • max_retries: Maximum number of retry attempts
  • self.retry(): Raises a retry exception
  • countdown: Seconds to wait before retrying

Retry with Exponential Backoff

Automatic Exponential Backoff

@app.task(bind=True, max_retries=5, default_retry_delay=60)
def fetch_data(self, url):
    try:
        response = requests.get(url, timeout=10)
        response.raise_for_status()
        return response.json()
    except requests.RequestException as exc:
        # Exponential backoff: 60s, 120s, 240s, 480s, 960s
        raise self.retry(
            exc=exc,
            countdown=60 * (2 ** self.request.retries)
        )

Using retry_backoff Parameter

@app.task(
    bind=True,
    max_retries=5,
    retry_backoff=True,  # Enable exponential backoff
    retry_backoff_max=600,  # Max wait time (10 minutes)
    retry_jitter=True  # Add randomness to prevent thundering herd
)
def api_call(self, endpoint):
    try:
        response = requests.get(endpoint)
        response.raise_for_status()
        return response.json()
    except requests.RequestException as exc:
        raise self.retry(exc=exc)

Advanced Retry Strategies

Retry with Custom Logic

@app.task(bind=True, max_retries=10)
def process_payment(self, user_id, amount):
    try:
        # Attempt payment
        result = payment_gateway.charge(user_id, amount)
        return result
    except PaymentGatewayException as exc:
        # Don't retry for certain error types
        if exc.error_code == 'INSUFFICIENT_FUNDS':
            # Don't retry - permanent failure
            raise
        
        # Retry for transient errors
        if exc.error_code in ['NETWORK_ERROR', 'TIMEOUT', 'RATE_LIMIT']:
            # Calculate backoff based on error type
            if exc.error_code == 'RATE_LIMIT':
                countdown = 300  # Wait 5 minutes for rate limit
            else:
                countdown = 60 * (2 ** self.request.retries)
            
            raise self.retry(exc=exc, countdown=countdown)
        
        # Unknown error - retry with default backoff
        raise self.retry(exc=exc)

Retry with Different Exceptions

@app.task(bind=True, max_retries=5)
def complex_operation(self, data):
    try:
        # Step 1: Validate data
        validate_data(data)
        
        # Step 2: Process data
        result = process_data(data)
        
        # Step 3: Save to database
        save_to_db(result)
        
        return result
        
    except ValidationError as exc:
        # Don't retry validation errors
        logger.error(f"Validation failed: {exc}")
        raise
        
    except DatabaseConnectionError as exc:
        # Retry database connection errors
        raise self.retry(exc=exc, countdown=30)
        
    except ProcessingError as exc:
        # Retry processing errors with longer delay
        raise self.retry(exc=exc, countdown=120)
        
    except Exception as exc:
        # Catch-all for unexpected errors
        logger.error(f"Unexpected error: {exc}", exc_info=True)
        raise self.retry(exc=exc)

Retry Configuration Options

Task-Level Configuration

@app.task(
    bind=True,
    max_retries=5,                    # Maximum retry attempts
    default_retry_delay=60,           # Default delay in seconds
    retry_backoff=True,               # Enable exponential backoff
    retry_backoff_max=600,            # Maximum backoff time
    retry_jitter=True,                # Add randomness to backoff
    autoretry_for=(Exception,),       # Auto-retry for these exceptions
    retry_kwargs={'max_retries': 3}   # Override max_retries in retry()
)
def my_task(self):
    # Task implementation
    pass

Global Configuration

# In celery configuration
app.conf.task_default_max_retries = 3
app.conf.task_default_retry_delay = 60
app.conf.task_acks_late = True  # Acknowledge after task completion
app.conf.task_reject_on_worker_lost = True  # Retry if worker dies

Auto-Retry for Specific Exceptions

Using autoretry_for

@app.task(
    bind=True,
    autoretry_for=(ConnectionError, TimeoutError),
    max_retries=5,
    retry_backoff=True
)
def fetch_external_data(self, url):
    # Automatically retries on ConnectionError or TimeoutError
    response = requests.get(url, timeout=10)
    return response.json()

Custom Auto-Retry Logic

from celery.exceptions import Retry

@app.task(bind=True, max_retries=5)
def smart_retry_task(self, data):
    try:
        return process(data)
    except RetryableError as exc:
        # Check if we should retry based on error details
        if exc.should_retry():
            raise self.retry(
                exc=exc,
                countdown=exc.get_retry_delay()
            )
        else:
            # Permanent failure - don't retry
            raise

Retry with State Tracking

Tracking Retry Attempts

@app.task(bind=True, max_retries=5)
def tracked_retry_task(self, task_id):
    current_retry = self.request.retries
    
    # Log retry attempt
    logger.info(f"Task {task_id} - Attempt {current_retry + 1}/{self.max_retries}")
    
    try:
        result = perform_operation()
        
        # Update state on success
        update_task_state(task_id, 'completed', result)
        return result
        
    except Exception as exc:
        # Update state on failure
        update_task_state(task_id, 'retrying', {
            'attempt': current_retry + 1,
            'error': str(exc)
        })
        
        if current_retry < self.max_retries:
            raise self.retry(exc=exc, countdown=60 * (2 ** current_retry))
        else:
            # Max retries reached
            update_task_state(task_id, 'failed', {'error': str(exc)})
            raise

Retry with Circuit Breaker Pattern

Implementing Circuit Breaker

from celery import Celery
import redis

app = Celery('tasks', broker='redis://localhost:6379/0')
redis_client = redis.Redis()

@app.task(bind=True, max_retries=5)
def circuit_breaker_task(self, service_name):
    # Check circuit breaker state
    circuit_key = f"circuit:{service_name}"
    circuit_state = redis_client.get(circuit_key)
    
    if circuit_state == b'open':
        # Circuit is open - don't retry, fail fast
        raise Exception(f"Circuit breaker is open for {service_name}")
    
    try:
        result = call_external_service(service_name)
        
        # Success - reset failure count
        redis_client.delete(f"failures:{service_name}")
        return result
        
    except Exception as exc:
        # Increment failure count
        failures = redis_client.incr(f"failures:{service_name}")
        redis_client.expire(f"failures:{service_name}", 60)
        
        # Open circuit if too many failures
        if failures >= 5:
            redis_client.setex(circuit_key, 300, 'open')  # Open for 5 minutes
            raise Exception(f"Circuit breaker opened for {service_name}")
        
        # Retry with backoff
        raise self.retry(exc=exc, countdown=60 * (2 ** self.request.retries))

Retry Best Practices

1. Distinguish Between Retryable and Non-Retryable Errors

RETRYABLE_ERRORS = (
    ConnectionError,
    TimeoutError,
    TemporaryError,
    RateLimitError
)

NON_RETRYABLE_ERRORS = (
    ValidationError,
    AuthenticationError,
    PermissionError
)

@app.task(bind=True, max_retries=5)
def best_practice_task(self, data):
    try:
        return process(data)
    except NON_RETRYABLE_ERRORS:
        # Don't retry - log and fail
        logger.error("Non-retryable error occurred")
        raise
    except RETRYABLE_ERRORS as exc:
        # Retry with backoff
        raise self.retry(exc=exc, countdown=60)

2. Use Appropriate Retry Delays

@app.task(bind=True, max_retries=5)
def api_task(self, endpoint):
    try:
        return call_api(endpoint)
    except RateLimitError as exc:
        # Wait longer for rate limits
        raise self.retry(exc=exc, countdown=300)
    except TimeoutError as exc:
        # Shorter delay for timeouts
        raise self.retry(exc=exc, countdown=10)
    except ConnectionError as exc:
        # Medium delay for connection errors
        raise self.retry(exc=exc, countdown=60)

3. Log Retry Attempts

import logging
from celery.utils.log import get_task_logger

logger = get_task_logger(__name__)

@app.task(bind=True, max_retries=5)
def logged_retry_task(self, data):
    retry_count = self.request.retries
    
    if retry_count > 0:
        logger.warning(
            f"Retrying task (attempt {retry_count + 1}/{self.max_retries})"
        )
    
    try:
        return process(data)
    except Exception as exc:
        logger.error(
            f"Task failed (attempt {retry_count + 1}): {exc}",
            exc_info=True
        )
        raise self.retry(exc=exc)

4. Set Reasonable Max Retries

# For critical tasks - more retries
@app.task(bind=True, max_retries=10)
def critical_task(self):
    pass

# For non-critical tasks - fewer retries
@app.task(bind=True, max_retries=3)
def non_critical_task(self):
    pass

5. Use Retry Jitter to Prevent Thundering Herd

@app.task(
    bind=True,
    max_retries=5,
    retry_backoff=True,
    retry_jitter=True  # Adds randomness to prevent all tasks retrying at once
)
def jittered_retry_task(self):
    try:
        return process()
    except Exception as exc:
        raise self.retry(exc=exc)

Monitoring Retries

Track Retry Metrics

from celery.signals import task_retry, task_failure

@task_retry.connect
def on_task_retry(sender=None, task_id=None, reason=None, einfo=None, **kwargs):
    logger.warning(f"Task {task_id} retrying: {reason}")
    # Increment retry counter in monitoring system
    increment_metric('celery.task.retries', tags=[f'task:{sender.name}'])

@task_failure.connect
def on_task_failure(sender=None, task_id=None, exception=None, traceback=None, **kwargs):
    if sender.request.retries >= sender.max_retries:
        logger.error(f"Task {task_id} failed after {sender.max_retries} retries")
        # Alert on final failure
        send_alert(f"Task {sender.name} exhausted all retries")

Common Retry Patterns

Pattern 1: Simple Retry with Fixed Delay

@app.task(bind=True, max_retries=3)
def simple_retry(self):
    try:
        return operation()
    except Exception as exc:
        raise self.retry(exc=exc, countdown=60)  # Wait 60 seconds

Pattern 2: Exponential Backoff

@app.task(bind=True, max_retries=5, retry_backoff=True)
def exponential_retry(self):
    try:
        return operation()
    except Exception as exc:
        raise self.retry(exc=exc)  # Automatically uses exponential backoff

Pattern 3: Conditional Retry

@app.task(bind=True, max_retries=5)
def conditional_retry(self, data):
    try:
        return operation(data)
    except Exception as exc:
        # Only retry if condition is met
        if should_retry(exc, data):
            raise self.retry(exc=exc)
        else:
            raise  # Don't retry

Pattern 4: Retry with Custom Arguments

@app.task(bind=True, max_retries=5)
def retry_with_custom_args(self, url, timeout=10):
    try:
        return fetch(url, timeout=timeout)
    except TimeoutError as exc:
        # Retry with increased timeout
        new_timeout = timeout * 2
        raise self.retry(
            exc=exc,
            kwargs={'timeout': new_timeout},
            countdown=60
        )

Summary

Key points for implementing retries in Celery:

  1. Always use bind=True to access the task instance
  2. Set appropriate max_retries based on task criticality
  3. Use exponential backoff for transient errors
  4. Distinguish retryable vs non-retryable errors
  5. Log retry attempts for debugging
  6. Use retry_jitter to prevent thundering herd problems
  7. Monitor retry metrics to identify problematic tasks
  8. Configure task_acks_late=True to ensure retries on worker crashes

Interview angle

  • “How do you retry a Celery task?” - autoretry_for with retry_backoff and retry_jitter, plus max_retries, or call self.retry(exc=...) explicitly with bind=True. Jitter matters for the same reason it does in HTTP retries: synchronised retries create a thundering herd.
  • “Which exceptions should trigger a retry?” - transient ones only: network errors, timeouts, upstream 5xx. Retrying a ValueError from bad input just burns the retry budget and delays the dead-letter.
  • “What happens after max retries?” - the task fails. Without a dead-letter path that failure is effectively lost, so route exhausted tasks somewhere inspectable and alert on the rate.
  • “Why must retried tasks be idempotent?” - acks_late plus retries means a task can execute more than once, including after partial completion. Key the effect on a business identifier so a repeat is a no-op.
Previous
What is Celery?