nuh-temp opened a new issue #8622:
URL: https://github.com/apache/airflow/issues/8622
**Apache Airflow version**: 1.10.10
**Environment**:
- **OS** (e.g. from /etc/os-release): Red Hat Enterprise Linux Workstation
release 7.6 (Maipo)
- **Kernel** (e.g. `uname -a`): 3.10.0-957.21.3.el7.x86_64
- **Install tools**:
- **Others**:
**What happened**:
When LocalExecutor is used with multiple threads allowed there seems to be a
race condition in scheduling tasks when there are many tasks in a DAG.
For example:
1. Task A gets scheduled;
2. Task A completes with AirflowRescheduleException and reschedules at T1;
3. Executors process other tasks which are up for execution;
4. By the time T1 not all tasks are processed yet and status of task A is
not read form "event_buffer" in executor BUT the task is up for being scheduled
again already;
5. Scheduler considers task A for execution and changes it status to
"queued";
6. When it comes to execution task fails with :
```
2020-04-11 20:58:45,511 [taskinstance.py:1128] [41699] [ERROR] Executor
reports task instance <TaskInstance: A 2020-04-11 18:36:46.526849+00:00
[queued]> finished (success) although the task says its queued. Was the task
killed externally?
```
**What you expected to happen**:
Tasks should not fail. Which means task should get scheduled again until
result of its execution hasn't been read from event buffer.
**How to reproduce it**:
**Anything else we need to know**:
Following patch fixes the issue:
```
$ git diff
diff --git a/airflow/executors/base_executor.py
b/airflow/executors/base_executor.py
index b0c6979c2..59b4f33a4 100644
--- a/airflow/executors/base_executor.py
+++ b/airflow/executors/base_executor.py
@@ -231,6 +231,12 @@ class BaseExecutor(LoggingMixin):
return cleared_events
+ def has_event(self, task_instance):
+ for key in list(self.event_buffer.keys()):
+ if task_instance.key == key:
+ return True
+ return False
+
def execute_async(self,
key: TaskInstanceKeyType,
command: CommandType,
diff --git a/airflow/jobs/scheduler_job.py b/airflow/jobs/scheduler_job.py
index 353b90b6b..38b3271e6 100644
--- a/airflow/jobs/scheduler_job.py
+++ b/airflow/jobs/scheduler_job.py
@@ -1281,6 +1281,9 @@ class SchedulerJob(BaseJob):
num_tasks_in_executor += 1
continue
+ if self.executor.has_event(task_instance):
+ continue
+
executable_tis.append(task_instance)
open_slots -= 1
dag_concurrency_map[dag_id] += 1
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
