title: Airflow : Run a task when some upstream is skipped by shortcircuit
body: <p>I have a task that I'll call <code>final</code> that has multiple upstream connections. When one of the upstreams gets skipped by <code>ShortCircuitOperator</code> this task gets skipped as well. I don't want <code>final</code> task to get skipped as it has to report on DAG success.</p>
  
  <p>To avoid it getting skipped I used <code>trigger_rule='all_done'</code>, but it still gets skipped.</p>
  
  <p>If I use <code>BranchPythonOperator</code> instead of <code>ShortCircuitOperator</code> <code>final</code> task doesn't get skipped. It would seem like branching workflow could be a solution, even though not optimal, but now <code>final</code> will not respect failures of upstream tasks.</p>
  
  <p>How do I get it to only run when upstreams are successful or skipped?</p>
  
  <p>Sample ShortCircuit DAG:</p>
  
  <pre class="lang-py prettyprint-override"><code>from airflow import DAG
  from airflow.operators.dummy_operator import DummyOperator
  from airflow.operators.python_operator import ShortCircuitOperator
  from datetime import datetime
  from random import randint
  
  default_args = {
      'owner': 'airflow',
      'start_date': datetime(2018, 8, 1)}
  
  dag = DAG(
      'shortcircuit_test',
      default_args=default_args,
      schedule_interval='* * * * *',
      catchup=False)
  
  def shortcircuit_fn():
      return randint(0, 1) == 1
  
  task_1 = DummyOperator(dag=dag, task_id='task_1')
  task_2 = DummyOperator(dag=dag, task_id='task_2')
  
  work = DummyOperator(dag=dag, task_id='work')
  short = ShortCircuitOperator(dag=dag, task_id='short_circuit', python_callable=shortcircuit_fn)
  final = DummyOperator(dag=dag, task_id="final", trigger_rule="all_done")
  
  task_1 &gt;&gt; short &gt;&gt; work &gt;&gt; final
  task_1 &gt;&gt; task_2 &gt;&gt; final
  </code></pre>
  
  <p><a href="https://i.stack.imgur.com/nqVff.png" rel="noreferrer"><img src="https://i.stack.imgur.com/nqVff.png" alt="DAG with shortcircuit operator"></a></p>
  
  <p>Sample Branch DAG:</p>
  
  <pre class="lang-py prettyprint-override"><code>from airflow import DAG
  from airflow.operators.dummy_operator import DummyOperator
  from airflow.operators.python_operator import BranchPythonOperator
  from datetime import datetime
  from random import randint
  
  default_args = {
      'owner': 'airflow',
      'start_date': datetime(2018, 8, 1)}
  
  dag = DAG(
      'branch_test',
      default_args=default_args,
      schedule_interval='* * * * *',
      catchup=False)
  
  # these two are only here to protect tasks from getting skipped as direct dependencies of branch operator
  to_do_work = DummyOperator(dag=dag, task_id='to_do_work')
  to_skip_work = DummyOperator(dag=dag, task_id='to_skip_work')
  
  def branch_fn():
      return to_do_work.task_id if randint(0, 1) == 1 else to_skip_work.task_id
  
  task_1 = DummyOperator(dag=dag, task_id='task_1')
  task_2 = DummyOperator(dag=dag, task_id='task_2')
  
  work = DummyOperator(dag=dag, task_id='work')
  branch = BranchPythonOperator(dag=dag, task_id='branch', python_callable=branch_fn)
  final = DummyOperator(dag=dag, task_id="final", trigger_rule="all_done")
  
  task_1 &gt;&gt; branch &gt;&gt; to_do_work &gt;&gt; work &gt;&gt; final
  branch &gt;&gt; to_skip_work &gt;&gt; final
  task_1 &gt;&gt; task_2 &gt;&gt; final
  </code></pre>
  
  <p><a href="https://i.stack.imgur.com/ioKIl.png" rel="noreferrer"><img src="https://i.stack.imgur.com/ioKIl.png" alt="DAG with branch operator"></a></p>