RabbitMQ JSON schema validation error discarding inference messages after model update

RabbitMQ JSON Schema Validation Error Discarding Inference Messages After Model Update

Problem

After deploying a new version of an AI inference model, the downstream service that consumes inference requests from a RabbitMQ queue started discarding messages. The messages are either silently dropped or routed to a dead‑letter exchange (DLX). Operators observed a sudden increase in the dead‑letter queue depth and a corresponding drop in inference throughput.

Typical symptoms:

• Consumer logs contain entries such as Message rejected: schema validation failed.
• RabbitMQ management UI shows a high rate of basic.reject with requeue=false.
• Dead‑letter queue inference.dlq accumulates JSON payloads that look correct at a glance.
• No network errors; the issue appears only after the model artifact was updated.

Root Cause

The new model version introduced additional fields to the inference request payload (e.g., model_version and request_id). The consumer application validates incoming messages against a static JSON schema that was bundled with the previous model release. Because the schema was not regenerated, every message containing the new fields fails validation, triggering the reject path in the RabbitMQ client library. In an air‑gapped environment the schema cannot be fetched from an external service, so the outdated schema persisted in the container image.

Key points of failure:

1. Schema version mismatch: The producer sends a payload that conforms to schema_v2.json, but the consumer validates against schema_v1.json.
2. Hard‑coded schema path: The consumer loads the schema from /app/schemas/inference_request.json at startup; the file was not updated during the model rollout.
3. Reject‑on‑validation policy: The client library is configured with basicReject(requeue=false) for any validation error, causing immediate dead‑lettering.

Debug & Investigation

Below is a step‑by‑step investigation that reproduced the issue.

1. Examine consumer logs

2024-10-22 14:03:12,874 ERROR [consumer] Message rejected: schema validation failed
Traceback (most recent call last):
  File "/app/consumer.py", line 112, in on_message
    validator.validate(message_body)
  File "/usr/local/lib/python3.9/site-packages/jsonschema/validators.py", line 215, in validate
    raise ValidationError(message)
jsonschema.exceptions.ValidationError: 'model_version' is a required property

2. Inspect the dead‑letter queue payload

{
  "input": {"text": "What is the capital of France?"},
  "model_version": "v2.1.0",
  "request_id": "c3f5e9b2-7a1d-4d3f-9f2a-8b6c1e5d9a4f"
}

3. Compare the two schema files

{
  "type": "object",
  "properties": {
    "input": {"type": "object"},
    "request_id": {"type": "string"}
  },
  "required": ["input", "request_id"]
}

{
  "type": "object",
  "properties": {
    "input": {"type": "object"},
    "request_id": {"type": "string"},
    "model_version": {"type": "string"}
  },
  "required": ["input", "request_id", "model_version"]
}

4. Verify RabbitMQ client configuration

channel.basic_consume(
    queue='inference.requests',
    on_message_callback=on_message,
    auto_ack=False
)
...
def on_message(ch, method, properties, body):
    try:
        validator.validate(json.loads(body))
        process_inference(json.loads(body))
        ch.basic_ack(delivery_tag=method.delivery_tag)
    except jsonschema.ValidationError:
        ch.basic_reject(delivery_tag=method.delivery_tag, requeue=False)

5. Confirm no external schema service is reachable

$ curl -s http://schema-registry.internal/v2/schema
curl: (7) Failed to connect to schema-registry.internal port 80: Connection refused

Solution

Update the consumer to load the correct schema version that matches the producer, and make the schema loading process resilient to future model updates.

1. Bundle the new schema with the container image

Replace /app/schemas/inference_request.json with the updated schema_v2.json and rename it to inference_request.json (the path expected by the code).

2. Modify the consumer to select the schema based on a version header

# BEFORE
SCHEMA_PATH = "/app/schemas/inference_request.json"
validator = jsonschema.Draft7Validator(load_schema(SCHEMA_PATH))

# AFTER
def load_schema_for_version(version: str):
    schema_map = {
        "v1": "/app/schemas/schema_v1.json",
        "v2": "/app/schemas/schema_v2.json"
    }
    path = schema_map.get(version, "/app/schemas/schema_v2.json")
    with open(path) as f:
        return json.load(f)

def on_message(ch, method, properties, body):
    payload = json.loads(body)
    version = payload.get("model_version", "v1")
    validator = jsonschema.Draft7Validator(load_schema_for_version(version))
    try:
        validator.validate(payload)
        process_inference(payload)
        ch.basic_ack(delivery_tag=method.delivery_tag)
    except jsonschema.ValidationError as e:
        logger.error("Message rejected: %s", e.message)
        ch.basic_reject(delivery_tag=method.delivery_tag, requeue=False)

3. Adjust the reject policy to allow requeue for transient validation errors

# OPTIONAL: only dead‑letter on permanent schema mismatch
if e.message.startswith("'model_version' is a required property"):
    ch.basic_reject(delivery_tag=method.delivery_tag, requeue=False)
else:
    ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)

4. Rebuild and redeploy the consumer image

$ docker build -t inference-consumer:2.1.0 .
$ kubectl set image deployment/inference-consumer inference-consumer=inference-consumer:2.1.0

Verify

After redeployment, perform the following checks:

1. Consume a test message that includes the new field:

   
   $ rabbitmqadmin publish exchange=inference.exchange routing_key=inference.requests payload='{
     "input": {"text":"test"},
     "request_id":"12345",
     "model_version":"v2.1.0"
   }'
   

   The consumer should log Message processed and acknowledge the delivery.

2. Inspect the dead‑letter queue:

   
   $ rabbitmqadmin get queue=inference.dlq requeue=false count=10
   # Expected: no messages returned
   

3. Metrics validation (if Prometheus exporter is enabled):

   
   # HELP rabbitmq_queue_messages_ready Number of ready messages in the queue
   # TYPE rabbitmq_queue_messages_ready gauge
   rabbitmq_queue_messages_ready{queue="inference.requests"} 0
   rabbitmq_queue_messages_ready{queue="inference.dlq"} 0
   

Prevent

Implement safeguards to avoid similar mismatches in the future.

• Schema versioning strategy: Store schemas in a version‑controlled directory and expose the version via a message header (e.g., x-schema-version) instead of inferring from payload.
• Automated CI check: Add a test that publishes a sample payload for each supported model version and asserts successful validation.
• Health‑check endpoint that reports the currently loaded schema version:

  
  GET /health/schema
  {
    "loaded_schema_version": "v2"
  }
  

• Fail‑fast deployment guard: In the CI/CD pipeline, compare the model_version field in the new model artifact against the schema version bundled in the consumer image. Abort if they diverge.
• Dead‑letter alerting: Configure an alert when the DLQ depth exceeds a threshold (e.g., >10 messages) for more than 5 minutes.

FAQ

1. Why does the consumer reject messages even though the JSON looks valid?
   Because validation is performed against a static schema that does not include the newly added model_version property. The JSON parser sees the extra field as a violation of the required list.
2. Can I disable schema validation to avoid this problem?
   Disabling validation removes a critical safety net. Instead, make the validation logic dynamic and version‑aware, as shown in the solution.
3. How do I handle backward compatibility when older producers still send v1 payloads?
   Maintain both schema_v1.json and schema_v2.json in the consumer image and select the appropriate schema based on the model_version field or a dedicated header.
4. What if the dead‑letter queue fills up and blocks the main queue?
   Configure a max‑length policy on the DLQ and set up a TTL with a dead‑letter back‑to‑origin to prevent back‑pressure. Additionally, monitor DLQ size and trigger scaling or manual purge actions.
5. Is there a way to test schema compatibility before deploying a new model?
   Yes. Include a pre‑deployment integration test that loads the new model artifact, generates a sample request, and runs the consumer’s validation routine against the bundled schema. Fail the pipeline if validation errors are detected.

Related Topic Hub: Data Infrastructure Troubleshooting Hub