[
https://issues.apache.org/jira/browse/AIRFLOW-928?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15925006#comment-15925006
]
Alex Guziel commented on AIRFLOW-928:
-------------------------------------
[~bolke] Did see a double trigger one hour after, will see if related.
> Same {task,execution_date} run multiple times in worker when using
> CeleryExecutor
> ---------------------------------------------------------------------------------
>
> Key: AIRFLOW-928
> URL: https://issues.apache.org/jira/browse/AIRFLOW-928
> Project: Apache Airflow
> Issue Type: Bug
> Components: celery
> Affects Versions: Airflow 1.7.1.3
> Environment: Docker
> Reporter: Uri Shamay
> Attachments: airflow.log, dag_runs.png, dummy_dag.py, processes.list,
> rabbitmq.queue, scheduler.log, worker_2.log, worker.log
>
>
> Hi,
> When using with Airflow with CeleryExecutor, both RabbitMQ && Redis I tested,
> I see that when workers are down, the scheduler run each period of time
> **append** to the same key of {task,execution_date} in the broker, the same
> {task,execution_date}, what means is that if workers are down/can't connect
> to broker for few hours, I got in the broker thousands of same executions.
> In my scenario I have just one dummy dag to run with dag_concurrency of 4,
> I expected in that scenario that broker will hold just 4 messages, and the
> scheduler shouldn't queuing another and another and another for same {task,
> execution_date}
> What happened is that when workers start to consume messages, they got
> thousands of tasks for just 4 tasks, and when they trying to write to
> database for task_instances - there are errors of integrity while such
> {task,execution_date} already exist.
> Note that in my test after let Airflow to consume works of just one dag
> without workers for few hours, then I connect to the broker outside by custom
> client and retrieve the messages - there was thousands of same
> {dag,execution_date}.
> Even if the case is that there are a lot of dag works on the same key that
> run just one instance when poll thousands - it's still bad behavior, better
> to produce one message to the queue, and if some timeout occurred (like
> visibility), to set the key - and not append to it.
> What happened is when workers are down for long time and have a lot of jobs
> that scheduled each minute, when workers come back, they got thousands of
> same jobs => cause to the worker to run the same dags a lot of times => a lot
> of wasted python runners => utilized all celery worker threads/processes =>
> starve all other jobs till he understood that need just one instance from all
> same.
> Attached files:
> 1. airflow.log - this is the task log, you can see that few instances
> processes of same {task,execution_date} write to the same log file.
> 2. worker.log - this is the worker log, you can see that worker trying to run
> same {task,execution_date} multiple times + the errors from the database
> integrity that said that those tasks on those dates already exists.
> 3. scheduler.log to show that scheduler decided to send again and again and
> again infinitely the same {job,execution_date}
> 4. the dummy_dag.py of the test
> 5. rabbitmq.queue - show that after 5 minutes the broker queue contains 40
> messages of same 4 {job,execution_date}
> 6. dag_runs.png - show that there are only 4 jobs that need to be run, while
> there are much more messages in the queue
> 7. processes.list - show that when start worker and doing: ps -ef | grep
> "airflow run", it show that worker run multiple times same
> {job,execution_date}
> 8. worker_2.log - show that when worker started - the same
> {job,execution_date} keys shown multiple times
> Thanks.
--
This message was sent by Atlassian JIRA
(v6.3.15#6346)
