78 lines
2.2 KiB
Python
78 lines
2.2 KiB
Python
import json
|
|
from datetime import datetime, timezone
|
|
from typing import Any, Dict, Optional
|
|
|
|
from app.api.producer import MessageProducer
|
|
from app.common.config import config
|
|
from app.common.logging import setup_logger
|
|
from app.worker.metrics import worker_metrics
|
|
|
|
logger = setup_logger("processing-worker")
|
|
|
|
ALLOWED_FAILURE_REASONS = {
|
|
"missing_required_field",
|
|
"invalid_rat_value",
|
|
"invalid_epoch",
|
|
"malformed_json",
|
|
"normalization_error",
|
|
}
|
|
|
|
|
|
def build_dlq_envelope(
|
|
original_payload: Any,
|
|
failure_reason: str,
|
|
failed_field: Optional[str] = None,
|
|
worker_instance: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Constructs a standardized DLQ envelope according to specification.
|
|
"""
|
|
if failure_reason not in ALLOWED_FAILURE_REASONS:
|
|
raise ValueError(f"Unsupported DLQ reason: {failure_reason}")
|
|
return {
|
|
"original_payload": original_payload,
|
|
"failure_reason": failure_reason,
|
|
"failed_field": failed_field,
|
|
"failed_at": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
|
|
"worker_instance": worker_instance or config.worker.instance_id,
|
|
}
|
|
|
|
|
|
def route_to_dlq(
|
|
producer: MessageProducer,
|
|
topic: str,
|
|
original_payload: Any,
|
|
failure_reason: str,
|
|
failed_field: Optional[str] = None,
|
|
raw_id: Optional[str] = None,
|
|
) -> bool:
|
|
"""
|
|
Wraps unprocessable record in a DLQ envelope, updates DLQ metric,
|
|
and publishes to the DLQ topic.
|
|
"""
|
|
envelope = build_dlq_envelope(
|
|
original_payload=original_payload,
|
|
failure_reason=failure_reason,
|
|
failed_field=failed_field,
|
|
)
|
|
envelope_bytes = json.dumps(envelope, default=str).encode("utf-8")
|
|
|
|
safe_key = None
|
|
if raw_id is not None:
|
|
try:
|
|
raw_id.encode("utf-8")
|
|
safe_key = raw_id
|
|
except UnicodeError:
|
|
safe_key = None
|
|
|
|
producer.produce(topic=topic, key=safe_key, value=envelope_bytes).result(
|
|
timeout=config.kafka.delivery_timeout_seconds + 1
|
|
)
|
|
worker_metrics.record_dlq(failure_reason, 1)
|
|
|
|
logger.warning(
|
|
f"Record routed to DLQ: reason={failure_reason}, field={failed_field}",
|
|
extra={"raw_id": safe_key, "event": "routed_to_dlq"},
|
|
)
|
|
return True
|