<p dir="auto"><strong>Apache Airflow version</strong>: 1.10.12</p>
<p dir="auto"><strong>Environment</strong>:</p>
<ul dir="auto">
<li><strong>Cloud provider or hardware configuration</strong>: Amazon EC2 instance, 4 CPU cores, 8GB RAM</li>
<li><strong>OS</strong> (e.g. from /etc/os-release): Ubuntu 18.04.5 LTS (Bionic Beaver)</li>
<li><strong>Kernel</strong> (e.g. <code class="notranslate">uname -a</code>): Linux ip-XX-XX-XX-XX.ec2.internal 5.4.0-1025-aws <a class="issue-link js-issue-link" data-error-text="Failed to load title" data-id="88074834" data-permission-text="Title is private" data-url="https://github.com/apache/airflow/issues/25" data-hovercard-type="pull_request" data-hovercard-url="/apache/airflow/pull/25/hovercard" href="https://github.com/apache/airflow/pull/25">#25</a>~18.04.1-Ubuntu SMP Fri Sep 11 12:03:04 UTC 2020 x86_64 x86_64 x86_64 GNU/Linux</li>
<li><strong>Install tools</strong>: Ansible Airflow role (<a href="https://github.com/idealista/airflow-role">https://github.com/idealista/airflow-role</a>)</li>
</ul>
<p dir="auto"><strong>What happened</strong>:</p>
<p dir="auto">I've made a combination of example DAGs from Airflow documentation (Branching section and Trigger Rules section):</p>
<details><summary>branching_dag.py</summary>
<div class="highlight highlight-source-python notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="from airflow.models import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.bash_operator import BashOperator
from airflow.operators.python_operator import BranchPythonOperator
from airflow.utils.dates import days_ago
from airflow.utils.trigger_rule import TriggerRule

dag = DAG(
    dag_id='branching_dag',
    schedule_interval=None,
    start_date=days_ago(1)
)

run_this_first = DummyOperator(task_id='run_this_first', dag=dag)

branching = BranchPythonOperator(
    task_id='branching', dag=dag,
    python_callable=lambda: 'branch_a'
)

branch_a = DummyOperator(task_id='branch_a', dag=dag)
follow_branch_a = BashOperator(task_id='follow_branch_a', dag=dag, bash_command='sleep 30s')

branch_b = DummyOperator(task_id='branch_b', dag=dag)

join = DummyOperator(task_id='join', dag=dag, trigger_rule=TriggerRule.NONE_FAILED_OR_SKIPPED)

run_this_first &gt;&gt; branching
branching &gt;&gt; branch_a &gt;&gt; follow_branch_a &gt;&gt; join
branching &gt;&gt; join
branching &gt;&gt; branch_b &gt;&gt; join"><pre class="notranslate"><span class="pl-k">from</span> <span class="pl-s1">airflow</span>.<span class="pl-s1">models</span> <span class="pl-k">import</span> <span class="pl-v">DAG</span>
<span class="pl-k">from</span> <span class="pl-s1">airflow</span>.<span class="pl-s1">operators</span>.<span class="pl-s1">dummy_operator</span> <span class="pl-k">import</span> <span class="pl-v">DummyOperator</span>
<span class="pl-k">from</span> <span class="pl-s1">airflow</span>.<span class="pl-s1">operators</span>.<span class="pl-s1">bash_operator</span> <span class="pl-k">import</span> <span class="pl-v">BashOperator</span>
<span class="pl-k">from</span> <span class="pl-s1">airflow</span>.<span class="pl-s1">operators</span>.<span class="pl-s1">python_operator</span> <span class="pl-k">import</span> <span class="pl-v">BranchPythonOperator</span>
<span class="pl-k">from</span> <span class="pl-s1">airflow</span>.<span class="pl-s1">utils</span>.<span class="pl-s1">dates</span> <span class="pl-k">import</span> <span class="pl-s1">days_ago</span>
<span class="pl-k">from</span> <span class="pl-s1">airflow</span>.<span class="pl-s1">utils</span>.<span class="pl-s1">trigger_rule</span> <span class="pl-k">import</span> <span class="pl-v">TriggerRule</span>

<span class="pl-s1">dag</span> <span class="pl-c1">=</span> <span class="pl-v">DAG</span>(
    <span class="pl-s1">dag_id</span><span class="pl-c1">=</span><span class="pl-s">'branching_dag'</span>,
    <span class="pl-s1">schedule_interval</span><span class="pl-c1">=</span><span class="pl-c1">None</span>,
    <span class="pl-s1">start_date</span><span class="pl-c1">=</span><span class="pl-en">days_ago</span>(<span class="pl-c1">1</span>)
)

<span class="pl-s1">run_this_first</span> <span class="pl-c1">=</span> <span class="pl-v">DummyOperator</span>(<span class="pl-s1">task_id</span><span class="pl-c1">=</span><span class="pl-s">'run_this_first'</span>, <span class="pl-s1">dag</span><span class="pl-c1">=</span><span class="pl-s1">dag</span>)

<span class="pl-s1">branching</span> <span class="pl-c1">=</span> <span class="pl-v">BranchPythonOperator</span>(
    <span class="pl-s1">task_id</span><span class="pl-c1">=</span><span class="pl-s">'branching'</span>, <span class="pl-s1">dag</span><span class="pl-c1">=</span><span class="pl-s1">dag</span>,
    <span class="pl-s1">python_callable</span><span class="pl-c1">=</span><span class="pl-k">lambda</span>: <span class="pl-s">'branch_a'</span>
)

<span class="pl-s1">branch_a</span> <span class="pl-c1">=</span> <span class="pl-v">DummyOperator</span>(<span class="pl-s1">task_id</span><span class="pl-c1">=</span><span class="pl-s">'branch_a'</span>, <span class="pl-s1">dag</span><span class="pl-c1">=</span><span class="pl-s1">dag</span>)
<span class="pl-s1">follow_branch_a</span> <span class="pl-c1">=</span> <span class="pl-v">BashOperator</span>(<span class="pl-s1">task_id</span><span class="pl-c1">=</span><span class="pl-s">'follow_branch_a'</span>, <span class="pl-s1">dag</span><span class="pl-c1">=</span><span class="pl-s1">dag</span>, <span class="pl-s1">bash_command</span><span class="pl-c1">=</span><span class="pl-s">'sleep 30s'</span>)

<span class="pl-s1">branch_b</span> <span class="pl-c1">=</span> <span class="pl-v">DummyOperator</span>(<span class="pl-s1">task_id</span><span class="pl-c1">=</span><span class="pl-s">'branch_b'</span>, <span class="pl-s1">dag</span><span class="pl-c1">=</span><span class="pl-s1">dag</span>)

<span class="pl-s1">join</span> <span class="pl-c1">=</span> <span class="pl-v">DummyOperator</span>(<span class="pl-s1">task_id</span><span class="pl-c1">=</span><span class="pl-s">'join'</span>, <span class="pl-s1">dag</span><span class="pl-c1">=</span><span class="pl-s1">dag</span>, <span class="pl-s1">trigger_rule</span><span class="pl-c1">=</span><span class="pl-v">TriggerRule</span>.<span class="pl-v">NONE_FAILED_OR_SKIPPED</span>)

<span class="pl-s1">run_this_first</span> <span class="pl-c1">&gt;&gt;</span> <span class="pl-s1">branching</span>
<span class="pl-s1">branching</span> <span class="pl-c1">&gt;&gt;</span> <span class="pl-s1">branch_a</span> <span class="pl-c1">&gt;&gt;</span> <span class="pl-s1">follow_branch_a</span> <span class="pl-c1">&gt;&gt;</span> <span class="pl-s1">join</span>
<span class="pl-s1">branching</span> <span class="pl-c1">&gt;&gt;</span> <span class="pl-s1">join</span>
<span class="pl-s1">branching</span> <span class="pl-c1">&gt;&gt;</span> <span class="pl-s1">branch_b</span> <span class="pl-c1">&gt;&gt;</span> <span class="pl-s1">join</span></pre></div>
</details>
<p dir="auto">Here is original image from Branching section of documentation:<br>
<a target="_blank" rel="noopener noreferrer nofollow" href="https://camo.githubusercontent.com/095598f70c036506ec251c55d76a2dc190855c811b0c9dc452e407a062b98a42/68747470733a2f2f616972666c6f772e6170616368652e6f72672f646f63732f737461626c652f5f696d616765732f6272616e63685f6e6f74652e706e67"><img src="https://camo.githubusercontent.com/095598f70c036506ec251c55d76a2dc190855c811b0c9dc452e407a062b98a42/68747470733a2f2f616972666c6f772e6170616368652e6f72672f646f63732f737461626c652f5f696d616765732f6272616e63685f6e6f74652e706e67" alt="original branching" data-canonical-src="https://airflow.apache.org/docs/stable/_images/branch_note.png" style="max-width: 100%;"></a><br>
and here is image from Trigger Rules section:<br>
<a target="_blank" rel="noopener noreferrer nofollow" href="https://camo.githubusercontent.com/fa3259be524e748330c543918bc4884b11edcda33d39ebbd66ef4fefeb4825d9/68747470733a2f2f616972666c6f772e6170616368652e6f72672f646f63732f737461626c652f5f696d616765732f6272616e63685f776974685f747269676765722e706e67"><img src="https://camo.githubusercontent.com/fa3259be524e748330c543918bc4884b11edcda33d39ebbd66ef4fefeb4825d9/68747470733a2f2f616972666c6f772e6170616368652e6f72672f646f63732f737461626c652f5f696d616765732f6272616e63685f776974685f747269676765722e706e67" alt="original trigger rules" data-canonical-src="https://airflow.apache.org/docs/stable/_images/branch_with_trigger.png" style="max-width: 100%;"></a></p>
<p dir="auto">According to DAG graph whichever branch would be selected, <code class="notranslate">join</code> task should be executed anyway. But I'm getting this:<br>
<a target="_blank" rel="noopener noreferrer nofollow" href="https://user-images.githubusercontent.com/2358132/95450254-b2750c80-096e-11eb-9e97-32131bad7d9c.png"><img src="https://user-images.githubusercontent.com/2358132/95450254-b2750c80-096e-11eb-9e97-32131bad7d9c.png" alt="airflow-branching-problem" style="max-width: 100%;"></a></p>
<p dir="auto">So, <code class="notranslate">join</code> task is skipped right after <code class="notranslate">branch_a</code> is selected.</p>
<p dir="auto"><strong>What you expected to happen</strong>:</p>
<p dir="auto">According to documentation (Branching section):</p>
<blockquote>
<p dir="auto">Paths of the branching task are branch_a, join and branch_b. Since join is a downstream task of branch_a, it will be excluded from the skipped tasks when branch_a is returned by the Python callable.</p>
</blockquote>
<p dir="auto">and from Trigger rules section:</p>
<blockquote>
<p dir="auto">The join task will be triggered as soon as branch_false has been skipped (a valid completion state) and follow_branch_a has succeeded. Because skipped tasks will not cascade through none_failed_or_skipped.</p>
</blockquote>
<p dir="auto">But it's not fulfilled in practice.</p>
<p dir="auto"><strong>Apache Airflow version</strong>:</p>
<p dir="auto">1.10.12</p>
<p dir="auto"><strong>What happened</strong>:</p>
<p dir="auto">Seems that from 1.10.12 the behavior from BranchPythonOperator was reversed. A task after all branches would be excluded from the skipped tasks before but now it is skipped. This prevents empty branches.</p>
<p dir="auto"><strong>What you expected to happen</strong>:</p>
<p dir="auto">Tasks after all branches should respect the trigger_rule and not be automatically skipped by the branch operator</p>
<p dir="auto"><strong>How to reproduce it</strong>:</p>
<p dir="auto">Run following code in a dag:</p>
<div class="highlight highlight-source-python notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="    def needs_some_extra_task(some_bool_field, **kwargs):
        if some_bool_field:
            return f&quot;extra_task&quot;
        else:
            return f&quot;final_task&quot;

    branch_op = BranchPythonOperator(
        dag=dag,
        task_id=f&quot;branch_task&quot;,
        provide_context=True,
        python_callable=needs_some_extra_task,
        op_kwargs={&quot;some_bool_field&quot;: True},  # For purposes of showing the problem
    )

    # will be always ran in this example
    extra_op = DummyOperator(
        dag=dag,
        task_id=f&quot;extra_task&quot;,
    )
    extra_op.set_upstream(branch_op)

    # will be skipped
    final_op = DummyOperator(
        dag=dag,
        task_id=&quot;final_task&quot;,
        trigger_rule=&quot;none_failed_or_skipped&quot;,
    )
    final_op.set_upstream([extra_op, branch_op])
"><pre class="notranslate">    <span class="pl-k">def</span> <span class="pl-en">needs_some_extra_task</span>(<span class="pl-s1">some_bool_field</span>, <span class="pl-c1">**</span><span class="pl-s1">kwargs</span>):
        <span class="pl-k">if</span> <span class="pl-s1">some_bool_field</span>:
            <span class="pl-k">return</span> <span class="pl-s">f"extra_task"</span>
        <span class="pl-k">else</span>:
            <span class="pl-k">return</span> <span class="pl-s">f"final_task"</span>

    <span class="pl-s1">branch_op</span> <span class="pl-c1">=</span> <span class="pl-v">BranchPythonOperator</span>(
        <span class="pl-s1">dag</span><span class="pl-c1">=</span><span class="pl-s1">dag</span>,
        <span class="pl-s1">task_id</span><span class="pl-c1">=</span><span class="pl-s">f"branch_task"</span>,
        <span class="pl-s1">provide_context</span><span class="pl-c1">=</span><span class="pl-c1">True</span>,
        <span class="pl-s1">python_callable</span><span class="pl-c1">=</span><span class="pl-s1">needs_some_extra_task</span>,
        <span class="pl-s1">op_kwargs</span><span class="pl-c1">=</span>{<span class="pl-s">"some_bool_field"</span>: <span class="pl-c1">True</span>},  <span class="pl-c"># For purposes of showing the problem</span>
    )

    <span class="pl-c"># will be always ran in this example</span>
    <span class="pl-s1">extra_op</span> <span class="pl-c1">=</span> <span class="pl-v">DummyOperator</span>(
        <span class="pl-s1">dag</span><span class="pl-c1">=</span><span class="pl-s1">dag</span>,
        <span class="pl-s1">task_id</span><span class="pl-c1">=</span><span class="pl-s">f"extra_task"</span>,
    )
    <span class="pl-s1">extra_op</span>.<span class="pl-en">set_upstream</span>(<span class="pl-s1">branch_op</span>)

    <span class="pl-c"># will be skipped</span>
    <span class="pl-s1">final_op</span> <span class="pl-c1">=</span> <span class="pl-v">DummyOperator</span>(
        <span class="pl-s1">dag</span><span class="pl-c1">=</span><span class="pl-s1">dag</span>,
        <span class="pl-s1">task_id</span><span class="pl-c1">=</span><span class="pl-s">"final_task"</span>,
        <span class="pl-s1">trigger_rule</span><span class="pl-c1">=</span><span class="pl-s">"none_failed_or_skipped"</span>,
    )
    <span class="pl-s1">final_op</span>.<span class="pl-en">set_upstream</span>([<span class="pl-s1">extra_op</span>, <span class="pl-s1">branch_op</span>])</pre></div>
<p dir="auto"><strong>Temporary workaround</strong>:</p>
<p dir="auto">Add dummy operators to any branch with no tasks</p>