title: Airflow ExternalTaskSensor don't fail when External Task fails
body: <p>I was trying to use the <code>ExternalTaskSensor</code> in Airflow 1.10.11 to manage the coordinate some dags. I have develop this code to test the functionality:</p>
  <pre class="lang-py prettyprint-override"><code>import time
  from datetime import datetime, timedelta
  from pprint import pprint
  
  from airflow import DAG
  from airflow.operators.dagrun_operator import TriggerDagRunOperator
  from airflow.operators.dummy_operator import DummyOperator
  from airflow.operators.python_operator import PythonOperator
  from airflow.sensors.external_task_sensor import ExternalTaskSensor
  from airflow.utils.state import State
  
  sensors_dag = DAG(
      &quot;test_launch_sensors&quot;,
      schedule_interval=None,
      start_date=datetime(2020, 2, 14, 0, 0, 0),
      dagrun_timeout=timedelta(minutes=150),
      tags=[&quot;DEMO&quot;],
  )
  
  dummy_dag = DAG(
      &quot;test_dummy_dag&quot;,
      schedule_interval=None,
      start_date=datetime(2020, 2, 14, 0, 0, 0),
      dagrun_timeout=timedelta(minutes=150),
      tags=[&quot;DEMO&quot;],
  )
  
  
  def print_context(ds, **context):
      pprint(context['conf'])
  
  
  with dummy_dag:
      starts = DummyOperator(task_id=&quot;starts&quot;, dag=dummy_dag)
      empty = PythonOperator(
          task_id=&quot;empty&quot;,
          provide_context=True,
          python_callable=print_context,
          dag=dummy_dag,
      )
      ends = DummyOperator(task_id=&quot;ends&quot;, dag=dummy_dag)
  
      starts &gt;&gt; empty &gt;&gt; ends
  
  with sensors_dag:
      trigger = TriggerDagRunOperator(
          task_id=f&quot;trigger_{dummy_dag.dag_id}&quot;,
          trigger_dag_id=dummy_dag.dag_id,
          conf={&quot;key&quot;: &quot;value&quot;},
          execution_date=&quot;{{ execution_date }}&quot;,
      )
      sensor = ExternalTaskSensor(
          task_id=&quot;wait_for_dag&quot;,
          external_dag_id=dummy_dag.dag_id,
          external_task_id=&quot;ends&quot;,
          failed_states=[&quot;failed&quot;, &quot;upstream_failed&quot;],
          poke_interval=5,
          timeout=120,
      )
      trigger &gt;&gt; sensor
  </code></pre>
  <p>The idea is that one dag triggers another one with a <code>TriggerDagRunOperator</code>. This sets the <code>execution_date</code> to the same value in both dags. This works perfectly when the state of the <code>dummy_dag</code> last task, <code>ends</code>, is <code>success</code>.</p>
  <p>However, if I force the intermediate task to fail like so:</p>
  <pre class="lang-py prettyprint-override"><code>def print_context(ds, **context):
      pprint(context['conf'])
      raise Exception('ouch')
  </code></pre>
  <p>The Sensor doesn't detect the <code>failed</code> or the <code>upstream_failed</code> states, and it keeps running until it times out. I was using the <code>failed_states</code> parameter to indicate which states need to be consider as failure, but it seems that is not working.</p>
  <p>Am I doing something wrong?</p>