title: Airflow BranchPythonOperator doesn't follow the specified branch
body: <p>I have an AIRFLOW DAG with the following structure.</p>
  
  <p><a href="https://i.stack.imgur.com/aCNtN.png" rel="nofollow noreferrer"><img src="https://i.stack.imgur.com/aCNtN.png" alt="Airflow Graph View"></a></p>
  
  <p>All the functions that start with "check*" are BranchPythonOperator, and the function exceptionControl is a ExecuteDagRunOperator that receives every error in order to handle them.</p>
  
  <p>This is the DAG configuration:</p>
  
  <pre><code>checkCloudFunctions = BranchPythonOperator(
      task_id='checkCloudFunctions',
      python_callable=check_cloud_functions,
      provide_context=True,
      dag=dag)
  
  checkSqlTables = BranchPythonOperator(
      task_id='checkSqlTables',
      python_callable=check_sql_tables,
      provide_context=True,
      dag=dag)
  
  checkBigQueryTable = BranchPythonOperator(
      task_id='checkBigQueryTable',
      python_callable=check_big_query_table,
      provide_context=True,
      dag=dag)
  
  labBuilt = DummyOperator(
      task_id='labBuilt',
      dag=dag)
  
  exceptionControl = ExecuteDagRunOperator(
      task_id='exceptionControl',
      execute_dag_id="SYS_exception_control",
      python_callable=mediation.dag_trigger_exception,
      trigger_rule='one_success',
      dag=dag)
  
  # graphs
  checkCloudFunctions &gt;&gt; checkSqlTables
  checkCloudFunctions &gt;&gt; exceptionControl
  
  checkSqlTables &gt;&gt; checkBigQueryTable
  checkSqlTables &gt;&gt; exceptionControl
  
  checkBigQueryTable &gt;&gt; labBuilt
  checkBigQueryTable &gt;&gt; exceptionControl
  </code></pre>
  
  <p><strong>The problem is</strong> that checkSqlTables should follow to exception control but it skips and the DAG ends. The function returns "exceptionControl" as we can see in the checkSqlTables log:</p>
  
  <pre><code>   {base_task_runner.py:98} INFO - {python_operator.py:90} INFO - Done. Returned value was: exceptionControl
     {base_task_runner.py:98} INFO - {python_operator.py:118} INFO - Following branch exceptionControl
     {base_task_runner.py:98} INFO - {python_operator.py:119} INFO - Marking other directly downstream tasks as skipped
     {base_task_runner.py:98} INFO - {python_operator.py:128} INFO - Done.
  </code></pre>
  
  <p>I also played with the <strong>trigger_rule</strong> attribute (one_success, dummy...) but it doesn't seems to work.</p>
  
  <p>If I delete the first step, it seems to work, so it seems it should be some kind of configuration problem with my dag.</p>
  
  <p><a href="https://i.stack.imgur.com/1q5Ly.png" rel="nofollow noreferrer"><img src="https://i.stack.imgur.com/1q5Ly.png" alt="enter image description here"></a></p>
  
  <p>Any ideas why the function checkSqlTables doesn't branch to exceptionControl?</p>
  
  <p>EDIT: In a <a href="https://airflow.apache.org/concepts.html#branching" rel="nofollow noreferrer">new deep reading to the Airflow Documentation</a> I noticed that if a step mark a task as skipped, it will be skipped forever, so my code will never work with Branching Operators. </p>
  
  <p>The solutions using branching consist on a dummy step before every step. But I have some DAG that have more than 10 steps and the schema will be completely a chaos.</p>