Job Failure Handling Patterns — Complete Guide
In this tutorial, you will learn about Job Failure Handling Patterns. We cover key concepts, practical examples, and best practices to help you master this topic.
Handle background job failures with dead letter queues, error classification, alerting, manual reprocessing, and graceful degradation strategies.
What You Learn
You will learn failure handling patterns: dead letter queues, error classification, automated alerts, manual reprocessing tools, and graceful degradation when services fail.
Why It Matters
Jobs fail. The difference between a reliable system and an unreliable one is how failures are handled. Proper failure handling ensures that no job is lost, operations teams are alerted, and the system degrades gracefully.
Real-World Use
DodaTech's failure handling system routes failed jobs to a dead letter queue, sends Slack alerts for critical failures, provides a web UI for manual reprocessing, and automatically retries at off-peak hours.
Dead Letter Queue
import redis
import json
import time
from datetime import datetime
r = redis.Redis()
class DeadLetterHandler:
def __init__(self, dlq_name='dead_letter'):
self.dlq_name = dlq_name
def send_to_dlq(self, job, error, retries):
entry = {
'job': job,
'error': str(error),
'retries': retries,
'failed_at': datetime.utcnow().isoformat(),
'status': 'failed',
}
r.lpush(self.dlq_name, json.dumps(entry))
print(f"Sent to DLQ: {job.get('task', 'unknown')} - {error}")
def get_failed_jobs(self, limit=10):
jobs = []
for i in range(min(limit, r.llen(self.dlq_name))):
data = r.lindex(self.dlq_name, i)
if data:
jobs.append(json.loads(data))
return jobs
def reprocess(self, index=0):
data = r.lindex(self.dlq_name, index)
if data:
entry = json.loads(data)
r.lpush('default', json.dumps(entry['job']))
print(f"Reprocessing: {entry['job'].get('task')}")
return True
return False
def clear_dlq(self):
r.delete(self.dlq_name)
print("DLQ cleared")
dlq = DeadLetterHandler()
dlq.send_to_dlq({'task': 'payment', 'amount': 100}, "Gateway timeout", 3)
dlq.send_to_dlq({'task': 'email', 'to': 'user@example.com'}, "SMTP error", 2)
failed = dlq.get_failed_jobs()
for f in failed:
print(f" {f['job']['task']}: {f['error']}")
dlq.reprocess(0)
Expected output:
Sent to DLQ: payment - Gateway timeout
Sent to DLQ: email - SMTP error
payment: Gateway timeout
email: SMTP error
Reprocessing: payment
Error Classification and Alerting
import logging
import json
from enum import Enum
class ErrorSeverity(Enum):
DEBUG = 1
WARNING = 2
CRITICAL = 3
class ErrorHandler:
def __init__(self):
self.logger = logging.getLogger(__name__)
def classify_error(self, exception):
if isinstance(exception, (ConnectionError, TimeoutError)):
return ErrorSeverity.WARNING
if isinstance(exception, (ValueError, TypeError)):
return ErrorSeverity.DEBUG
if isinstance(exception, (PermissionError, AuthenticationError)):
return ErrorSeverity.CRITICAL
return ErrorSeverity.WARNING
def handle(self, job, exception, retries):
severity = self.classify_error(exception)
error_info = {
'job': job,
'error': str(exception),
'type': type(exception).__name__,
'severity': severity.name,
'retries': retries,
}
if severity == ErrorSeverity.CRITICAL:
self.send_alert(error_info)
elif severity == ErrorSeverity.WARNING:
self.logger.warning(json.dumps(error_info))
else:
self.logger.debug(json.dumps(error_info))
if retries >= 3:
send_to_dlq(job, exception, retries)
elif severity == ErrorSeverity.CRITICAL:
send_to_dlq(job, exception, retries)
def send_alert(self, info):
print(f"ALERT: {info['error']} ({info['severity']})")
class AuthenticationError(Exception):
pass
def send_to_dlq(job, error, retries):
print(f" DLQ: {job.get('task')} after {retries} retries")
handler = ErrorHandler()
handler.handle({'task': 'payment'}, TimeoutError("Gateway timeout"), 2)
handler.handle({'task': 'auth'}, AuthenticationError("Invalid token"), 1)
handler.handle({'task': 'cleanup'}, ValueError("Bad data"), 5)
Expected output:
ALERT: Invalid token (CRITICAL)
DLQ: auth after 1 retries
DLQ: cleanup after 5 retries
Graceful Degradation
import random
class ServiceDegradation:
def __init__(self):
self.primary_available = True
self.fallback_count = 0
def process(self, data):
if self.primary_available and self.check_primary():
return self.primary_process(data)
return self.fallback_process(data)
def check_primary(self):
if random.random() < 0.3:
self.primary_available = False
return False
return True
def primary_process(self, data):
return f"Primary: {data['id']} processed"
def fallback_process(self, data):
self.fallback_count += 1
result = f"Fallback: {data['id']} queued"
if self.fallback_count > 5:
self.primary_available = True
self.fallback_count = 0
return result
degradation = ServiceDegradation()
for i in range(10):
result = degradation.process({'id': i})
print(result)
Expected output:
Primary: 0 processed
Primary: 1 processed
Fallback: 2 queued
Fallback: 3 queued
...
Manual Reprocessing UI
import redis
import json
from datetime import datetime
r = redis.Redis()
class ReprocessUI:
def list_failed(self, queue='dead_letter', limit=20):
jobs = []
for i in range(min(limit, r.llen(queue))):
data = r.lindex(queue, i)
if data:
jobs.append(json.loads(data))
return jobs
def reprocess_single(self, index, queue='dead_letter', target='default'):
data = r.lindex(queue, index)
if data:
entry = json.loads(data)
r.lpush(target, json.dumps(entry['job']))
r.lset(queue, index, json.dumps({**entry, 'status': 'reprocessing'}))
return True
return False
def reprocess_all(self, queue='dead_letter', target='default'):
count = 0
for i in range(r.llen(queue)):
if self.reprocess_single(i, queue, target):
count += 1
return count
def delete_failed(self, index, queue='dead_letter'):
job = r.lindex(queue, index)
if job:
entry = json.loads(job)
entry['status'] = 'deleted'
r.lset(queue, index, json.dumps(entry))
return True
return False
ui = ReprocessUI()
# Simulate failed jobs
r.lpush('dead_letter', json.dumps({'job': {'task': 'payment'}, 'error': 'timeout', 'status': 'failed'}))
r.lpush('dead_letter', json.dumps({'job': {'task': 'email'}, 'error': 'smtp', 'status': 'failed'}))
print("Failed jobs:")
for j in ui.list_failed():
print(f" {j['job']['task']}: {j['error']} [{j['status']}]")
count = ui.reprocess_all()
print(f"\nReprocessed {count} jobs")
Expected output:
Failed jobs:
email: smtp [failed]
payment: timeout [failed]
Reprocessed 2 jobs
Common Mistakes
1. No Dead Letter Queue
Without a DLQ, failed jobs disappear forever. You cannot inspect, debug, or reprocess them. Always route exhausted retries to a DLQ.
2. Not Classifying Errors
Treating all errors the same leads to retrying permanent failures. Classify errors by type and severity to determine the correct response.
3. Silent Failures
A failed job that only logs and continues is invisible. Always alert on critical failures. Monitor failure rates.
4. No Manual Reprocess
When a DLQ fills up, operations need a way to reprocess jobs. Build a reprocess UI or CLI that moves jobs from DLQ back to the main queue.
5. Not Testing Failure Scenarios
Failure handling code that has never been tested will fail when needed. Simulate failures in staging. Verify DLQ routing, alerts, and reprocessing.
Practice Questions
1. What is a dead letter queue?
A queue for jobs that have exhausted retries. Failed jobs are stored for manual inspection, debugging, and eventual reprocessing.
2. How do you classify errors?
By type (transient vs permanent), severity (debug, warning, critical), and source (network, data, auth, system).
3. What is graceful degradation?
When a primary service fails, the system uses a fallback Strategy that provides limited but functional service instead of failing completely.
4. How do you implement manual reprocessing?
A tool (CLI or UI) that lists failed jobs and allows moving them back to the main queue for reprocessing.
Challenge
Build a complete failure handling system for a payment service: classify errors (card declined=permanent, gateway timeout=transient, auth failure=critical), route to DLQ after 3 retries, send Slack alert for critical errors, provide CLI for reprocessing, and implement fallback to batch processing on primary failure.
FAQ
Mini Project: Failure Handling System
import redis
import json
import time
from datetime import datetime
from enum import Enum
r = redis.Redis()
class Severity(Enum):
LOW = 1
MEDIUM = 2
HIGH = 3
class FailureHandler:
def __init__(self):
self.alert_log = []
def handle(self, job, error, retries):
severity = self._classify(error)
record = {
'job': job,
'error': str(error),
'type': type(error).__name__,
'severity': severity.name,
'retries': retries,
'timestamp': datetime.utcnow().isoformat(),
}
dlq_name = f"dlq:{severity.name.lower()}"
r.lpush(dlq_name, json.dumps(record))
r.ltrim(dlq_name, 0, 9999)
if severity == Severity.HIGH:
self._alert(record)
self.alert_log.append(record)
return record
def _classify(self, error):
if isinstance(error, (ConnectionError, TimeoutError)):
return Severity.MEDIUM
if isinstance(error, (ValueError, KeyError)):
return Severity.LOW
if isinstance(error, PermissionError):
return Severity.HIGH
return Severity.MEDIUM
def _alert(self, record):
print(f"ALERT [{record['severity']}]: {record['type']} in {record['job'].get('task')}")
print(f" Error: {record['error']}")
def get_dlq_stats(self):
stats = {}
for sev in Severity:
name = f"dlq:{sev.name.lower()}"
stats[sev.name] = r.llen(name)
return stats
def reprocess_dlq(self, severity='medium', target='default'):
dlq_name = f"dlq:{severity}"
count = 0
for i in range(r.llen(dlq_name)):
data = r.lindex(dlq_name, 0)
if data:
entry = json.loads(data)
r.rpush(target, json.dumps(entry['job']))
r.lpop(dlq_name)
count += 1
return count
handler = FailureHandler()
handler.handle({'task': 'email'}, TimeoutError("SMTP timeout"), 2)
handler.handle({'task': 'report'}, ValueError("Bad data format"), 3)
handler.handle({'task': 'auth'}, PermissionError("Access denied"), 1)
print(f"\nDLQ Stats: {handler.get_dlq_stats()}")
reprocessed = handler.reprocess_dlq('medium')
print(f"Reprocessed {reprocessed} medium-severity jobs")
Expected output:
ALERT [HIGH]: PermissionError in auth
Error: Access denied
DLQ Stats: {'LOW': 1, 'MEDIUM': 1, 'HIGH': 1}
Reprocessed 1 medium-severity jobs
What's Next
Now that you understand failure handling, explore progress tracking for long-running jobs, then learn about job cancellation mechanisms.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro