[Python] Occasional 'NotImplementedError' in Airflow with DockerOperator

I'm using the CeleryExecutor and DockerOperator to run my tasks in Airflow. The docker containers are run on a DIND service running in the same K8s cluster as Airflow.

The DAGs are dynamically created using a Python file, i.e. they all have the same config and tasks, but with different parameters.

Interestingly, there is a particular DAG (let's call DAG X) which appears to frequently run into the following error when its tasks are started. The error is random and not consistent. Sometimes most of its tasks fail with the error, other times only a few. When you retry a task which has failed enough times, it succeeds.

[2025-02-14T09:59:35.485+0000] {taskinstance.py:2193} INFO - Starting attempt 1 of 1
[2025-02-14T09:59:35.507+0000] {taskinstance.py:2217} INFO - Executing <Task(DockerOperator): my_task> on 2025-02-14 09:56:38.717801+00:00
[2025-02-14T09:59:35.514+0000] {standard_task_runner.py:60} INFO - Started process 439369 to run task
[2025-02-14T09:59:35.517+0000] {standard_task_runner.py:87} INFO - Running: ['***', 'tasks', 'run', 'my_dag', 'my_task', 'manual__2025-02-14T09:56:38.717801+00:00', '--job-id', '105389', '--raw', '--subdir', 'DAGS_FOLDER/dynamic_dag.py', '--cfg-path', '/tmp/tmpre_1wgu6']
[2025-02-14T09:59:35.519+0000] {standard_task_runner.py:88} INFO - Job 105389: Subtask my_task
[2025-02-14T09:59:36.007+0000] {taskinstance.py:2731} ERROR - Task failed with exception
Traceback (most recent call last):
File "/home/airflow/.local/lib/python3.9/site-packages/airflow/models/taskinstance.py", line 444, in _execute_task
result = _execute_callable(context=context, **execute_callable_kwargs)
File "/home/airflow/.local/lib/python3.9/site-packages/airflow/models/taskinstance.py", line 414, in _execute_callable
return execute_callable(context=context, **execute_callable_kwargs)
File "/home/airflow/.local/lib/python3.9/site-packages/airflow/models/baseoperator.py", line 1175, in execute
raise NotImplementedError()
NotImplementedError
[2025-02-14T09:59:36.073+0000] {standard_task_runner.py:107} ERROR - Failed to execute job 105389 for task my_task (; 439369)
[2025-02-14T09:59:36.091+0000] {local_task_job_runner.py:234} INFO - Task exited with return code 1
[2025-02-14T09:59:36.153+0000] {taskinstance.py:3312} INFO - 0 downstream tasks scheduled from follow-on schedule check


The associated Celery task succeeds (based on the Flower monitoring tool), but Airflow raises the exception above, and I notice that no container was started on the DIND service. We also see the following log from the scheduler:

TaskInstance Finished: dag_id=my_dag, task_id=my_task, run_id=manual__2025-02-14T09:41:21.943302+00:00, map_index=-1, run_start_date=2025-02-14 09:41:26.398775+00:00, run_end_date=2025-02-14 09:41:27.013421+00:00, run_duration=0.614646, state=failed, executor_state=success, try_number=1, max_tries=0, job_id=105371, pool=default_pool, queue=default, priority_weight=10, operator=DockerOperator, queued_dttm=2025-02-14 09:41:23.067512+00:00, queued_by_job_id=97598, pid=438349


I can tell when the task is going to fail as there are no logs from docker.py, i.e. the first log indicates that the container is being started.

[2025-02-14, 09:56:42 UTC] {docker.py:486} INFO - Pulling docker image example/image:1.2.3


Note that this error has occurred with another DAG previously, but it appears to be very common with DAG X. Nothing sticks out when comparing the DAGs, especially as they are generated in the same way.


UPDATE: I just noticed that one task is defined as a Python task, and not DockerOperator. This task also suffers from the random NotImplementedError. So I suspect it is unrelated to the operator.

Continue reading...