BUG: AirflowException doesn't raise exception in SequentialExecutor or LocalExecutor

[jlowin]

Many bugs I've addressed recently share a symptom -- they all log an error that "The airflow run command failed at reporting an error. This should not occur in normal circumstances." (from https://github.com/airbnb/airflow/blob/master/airflow/jobs.py#L808)

Trying to reproduce. These are my notes...

Sequence of events:

1. Use SequentialExecutor, as with airflow test ...

2. The SequentialExecutor runs the task airflow command with (https://github.com/airbnb/airflow/blob/master/airflow/executors/sequential_executor.py):

   try:
       sp = subprocess.Popen(command, shell=True)
       sp.wait()
   except Exception as e:
       self.change_state(key, State.FAILED)
       raise e
   self.change_state(key, State.SUCCESS)

3. A task raises AirflowException prior to being executing

   • For example, the "dag not found" error raised by subdag doesn't work with LocalExecutor/SequentialExecutor #1168 (prior to fix in Handle both SubDagOperator classes #1196) due to SubDag errors

4. For some reason, the AirflowException is NOT caught by the executor's trap

   • pretty sure this is the problem, still exploring
   • don't we need to check sp.returncode rather than trapping errors?

5. The state gets changed to success

6. The task's state doesn't change

7. The "normal circumstances" error gets raised

[mistercrunch]

I think when there are kill -9 on a task instance you may get that as well.

[mistercrunch]

There's an obvious elif missing before this message is logged, it's if the task instance state in the DB is running and the executor reported failure, that a clear failure. The handle_failure method should be called at that point and the failure recorded.