Issue 1:
Apache Airflow version: 1.10.12

Environment:

Cloud provider or hardware configuration: Amazon EC2 instance, 4 CPU cores, 8GB RAM

OS (e.g. from /etc/os-release): Ubuntu 18.04.5 LTS (Bionic Beaver)

Kernel (e.g. uname -a): Linux ip-XX-XX-XX-XX.ec2.internal 5.4.0-1025-aws #25~18.04.1-Ubuntu SMP Fri Sep 11 12:03:04 UTC 2020 x86_64 x86_64 x86_64 GNU/Linux

Install tools: Ansible Airflow role (https://github.com/idealista/airflow-role)

What happened:

I've made a combination of example DAGs from Airflow documentation (Branching section and Trigger Rules section):

branching_dag.py

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 >> branching
branching >> branch_a >> follow_branch_a >> join
branching >> join
branching >> branch_b >> join

Here is original image from Branching section of documentation:

and here is image from Trigger Rules section:

According to DAG graph whichever branch would be selected, join task should be executed anyway. But I'm getting this:

So, join task is skipped right after branch_a is selected.

What you expected to happen:

According to documentation (Branching section):

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.

and from Trigger rules section:

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.

But it's not fulfilled in practice.

Issue 2:
Apache Airflow version:

1.10.12

What happened:

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.

What you expected to happen:

Tasks after all branches should respect the trigger_rule and not be automatically skipped by the branch operator

How to reproduce it:

Run following code in a dag:

def needs_some_extra_task(some_bool_field, **kwargs):
if some_bool_field:
return f"extra_task"
else:
return f"final_task"

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

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

# will be skipped
final_op = DummyOperator(
dag=dag,
task_id="final_task",
trigger_rule="none_failed_or_skipped",
)
final_op.set_upstream([extra_op, branch_op])

Temporary workaround:

Add dummy operators to any branch with no tasks