BacklogLimitExceeded - Airflow system with celery executor and rabbitmq

[Ali Naqvi <al...@conornash.com>]

Hi folks,
We have a Airflow-celery-rabbitmq system running a massive dag that has
2500 tasks, which runs into the following error after ~1000 messages are
sent to the rabbitmq

scheduler_1  | [2016-12-02 20:14:02,372] {jobs.py:758} ERROR -
f5ead71a-7c77-4287-bacd-9e06a8724855
scheduler_1  | Traceback (most recent call last):
scheduler_1  |   File
"/opt/pyenv/versions/3.5.1/lib/python3.5/site-packages/airflow/jobs.py",
line 755, in _execute
scheduler_1  |     executor.heartbeat()
scheduler_1  |   File
"/opt/pyenv/versions/3.5.1/lib/python3.5/site-packages/airflow/executors/base_executor.py",
line 107, in heartbeat
scheduler_1  |     self.sync()
scheduler_1  |   File
"/opt/pyenv/versions/3.5.1/lib/python3.5/site-packages/airflow/executors/celery_executor.py",
line 75, in sync
scheduler_1  |     state = async.state
scheduler_1  |   File
"/opt/pyenv/versions/3.5.1/lib/python3.5/site-packages/celery/result.py",
line 394, in state
scheduler_1  |     return self._get_task_meta()['status']
scheduler_1  |   File
"/opt/pyenv/versions/3.5.1/lib/python3.5/site-packages/celery/result.py",
line 339, in _get_task_meta
scheduler_1  |     return self._maybe_set_cache(self.backend.get_task_meta(
self.id))
scheduler_1  |   File
"/opt/pyenv/versions/3.5.1/lib/python3.5/site-packages/celery/backends/amqp.py",
line 180, in get_task_meta
scheduler_1  |     raise self.BacklogLimitExceeded(task_id)
scheduler_1  | celery.backends.amqp.BacklogLimitExceeded:
f5ead71a-7c77-4287-bacd-9e06a8724855
scheduler_1  | [2016-12-02 20:14:02,379] {jobs.py:759} ERROR - Tachycardia!

This seems to be a celery issue but we couldn't find any information
regarding this error for either Airflow or Celery.

Does anyone know how to solve this?

Ali