BashOperator to raise AirflowSkipException on exit code 127

airflow/example_dags/example_bash_operator.py

@@ -21,7 +21,7 @@
 from datetime import timedelta
 
 from airflow import DAG
-from airflow.operators.bash import BashOperator
+from airflow.operators.bash import EXIT_CODE_SKIP, BashOperator
 from airflow.operators.dummy import DummyOperator
 from airflow.utils.dates import days_ago
 
@@ -67,5 +67,14 @@
     # [END howto_operator_bash_template]
     also_run_this >> run_this_last
 
+# [START howto_operator_bash_skip]
+this_will_skip = BashOperator(
+    task_id='this_will_skip',
+    bash_command=f'echo "hello world"; exit {EXIT_CODE_SKIP};',
+    dag=dag,
+)
+# [END howto_operator_bash_skip]
+this_will_skip >> run_this_last
+
 if __name__ == "__main__":
     dag.cli()

airflow/operators/bash.py

@@ -23,11 +23,13 @@
 from tempfile import TemporaryDirectory, gettempdir
 from typing import Dict, Optional
 
-from airflow.exceptions import AirflowException
+from airflow.exceptions import AirflowException, AirflowSkipException
 from airflow.models import BaseOperator
 from airflow.utils.decorators import apply_defaults
 from airflow.utils.operator_helpers import context_to_airflow_vars
 
+EXIT_CODE_SKIP = 127
+
 
 class BashOperator(BaseOperator):
     r"""
@@ -51,16 +53,33 @@ class BashOperator(BaseOperator):
     :param output_encoding: Output encoding of bash command
     :type output_encoding: str
 
-    On execution of this operator the task will be up for retry
-    when exception is raised. However, if a sub-command exits with non-zero
-    value Airflow will not recognize it as failure unless the whole shell exits
-    with a failure. The easiest way of achieving this is to prefix the command
-    with ``set -e;``
-    Example:
+    Airflow will evaluate the exit code of the bash command.  In general, a non-zero exit code will result in
+    task failure and zero will result in task success.  Exit code ``255`` will throw an
+    :class:`airflow.exceptions.AirflowSkipException`, which will leave the task in ``skipped`` state.
 
-    .. code-block:: python
+    .. list-table::
+       :widths: 25 25
+       :header-rows: 1
+
+       * - Exit code range
+         - Behavior
+       * - 0
+         - success
+       * - 1-249
+         - raise :class:`airflow.exceptions.AirflowException`
+       * - 255
+         - raise :class:`airflow.exceptions.AirflowSkipException`
+
+    .. note::
+
+        Airflow will not recognize a non-zero exit code unless the whole shell exit with a non-zero exit
+        code.  This can be an issue if the non-zero exit arises from a sub-command.  The easiest way of
+        addressing this is to prefix the command with ``set -e;``
+
+        Example:
+        .. code-block:: python
 
-        bash_command = "set -e; python3 script.py '{{ next_execution_date }}'"
+            bash_command = "set -e; python3 script.py '{{ next_execution_date }}'"
 
     .. note::
 
@@ -176,7 +195,9 @@ def pre_exec():
 
             self.log.info('Command exited with return code %s', self.sub_process.returncode)
 
-            if self.sub_process.returncode != 0:
+            if self.sub_process.returncode == EXIT_CODE_SKIP:
+                raise AirflowSkipException(f"Bash returned exit code {EXIT_CODE_SKIP}. Skipping task")
+            elif self.sub_process.returncode != 0:
                 raise AirflowException('Bash command failed. The command returned a non-zero exit code.')
 
         return line

docs/apache-airflow/howto/operator/bash.rst

@@ -74,6 +74,18 @@ inside the bash_command, as below:
         env={'message': '{{ dag_run.conf["message"] if dag_run else "" }}'},
     )
 
+Skipping
+--------
+
+In general a non-zero exit code produces an AirflowException and thus a task failure.  In cases where it is desirable
+to instead have the task end in a ``skipped`` state, you can exit with code ``127``.
+
+.. exampleinclude:: /../../airflow/example_dags/example_bash_operator.py
+    :language: python
+    :start-after: [START howto_operator_bash_skip]
+    :end-before: [END howto_operator_bash_skip]
+
+
 Troubleshooting
 ---------------
 

tests/api/common/experimental/test_mark_tasks.py

@@ -47,7 +47,7 @@ def setUpClass(cls):
         models.DagBag(include_examples=True, read_dags_from_db=False).sync_to_db()
         dagbag = models.DagBag(include_examples=True, read_dags_from_db=True)
         dagbag.collect_dags_from_db()
-        cls.dag1 = dagbag.dags['example_bash_operator']
+        cls.dag1 = dagbag.dags['miscellaneous_test_dag']
         cls.dag1.sync_to_db()
         cls.dag2 = dagbag.dags['example_subdag_operator']
         cls.dag2.sync_to_db()
@@ -370,10 +370,19 @@ def test_mark_tasks_subdag(self):
 
 
 class TestMarkDAGRun(unittest.TestCase):
+    INITIAL_TASK_STATES = {
+        'runme_0': State.SUCCESS,
+        'runme_1': State.SKIPPED,
+        'runme_2': State.UP_FOR_RETRY,
+        'also_run_this': State.QUEUED,
+        'run_after_loop': State.RUNNING,
+        'run_this_last': State.FAILED,
+    }
+
     @classmethod
     def setUpClass(cls):
         dagbag = models.DagBag(include_examples=True, read_dags_from_db=False)
-        cls.dag1 = dagbag.dags['example_bash_operator']
+        cls.dag1 = dagbag.dags['miscellaneous_test_dag']
         cls.dag1.sync_to_db()
         cls.dag2 = dagbag.dags['example_subdag_operator']
         cls.dag2.sync_to_db()
@@ -382,27 +391,28 @@ def setUpClass(cls):
     def setUp(self):
         clear_db_runs()
 
+    def _get_num_tasks_with_starting_state(self, state: State, inclusion: bool):
+        """
+        If ``inclusion=True``, get num tasks with initial state ``state``.
+        Otherwise, get number tasks with initial state not equal to ``state``
+        :param state: State to compare against
+        :param inclusion: whether to look for inclusion or exclusion
+        :return: number of tasks meeting criteria
+        """
+        states = self.INITIAL_TASK_STATES.values()
+
+        def compare(x, y):
+            return x == y if inclusion else x != y
+
+        return len([s for s in states if compare(s, state)])
+
     def _set_default_task_instance_states(self, dr):
-        # success task
-        dr.get_task_instance('runme_0').set_state(State.SUCCESS)
-        # skipped task
-        dr.get_task_instance('runme_1').set_state(State.SKIPPED)
-        # retry task
-        dr.get_task_instance('runme_2').set_state(State.UP_FOR_RETRY)
-        # queued task
-        dr.get_task_instance('also_run_this').set_state(State.QUEUED)
-        # running task
-        dr.get_task_instance('run_after_loop').set_state(State.RUNNING)
-        # failed task
-        dr.get_task_instance('run_this_last').set_state(State.FAILED)
+        for task_id, state in self.INITIAL_TASK_STATES.items():
+            dr.get_task_instance(task_id).set_state(state)
 
     def _verify_task_instance_states_remain_default(self, dr):
-        assert dr.get_task_instance('runme_0').state == State.SUCCESS
-        assert dr.get_task_instance('runme_1').state == State.SKIPPED
-        assert dr.get_task_instance('runme_2').state == State.UP_FOR_RETRY
-        assert dr.get_task_instance('also_run_this').state == State.QUEUED
-        assert dr.get_task_instance('run_after_loop').state == State.RUNNING
-        assert dr.get_task_instance('run_this_last').state == State.FAILED
+        for task_id, state in self.INITIAL_TASK_STATES.items():
+            assert dr.get_task_instance(task_id).state == state
 
     @provide_session
     def _verify_task_instance_states(self, dag, date, state, session=None):
@@ -445,7 +455,8 @@ def test_set_running_dag_run_to_success(self):
         altered = set_dag_run_state_to_success(self.dag1, date, commit=True)
 
         # All except the SUCCESS task should be altered.
-        assert len(altered) == 5
+        expected = self._get_num_tasks_with_starting_state(State.SUCCESS, inclusion=False)
+        assert len(altered) == expected
         self._verify_dag_run_state(self.dag1, date, State.SUCCESS)
         self._verify_task_instance_states(self.dag1, date, State.SUCCESS)
         self._verify_dag_run_dates(self.dag1, date, State.SUCCESS, middle_time)
@@ -457,9 +468,9 @@ def test_set_running_dag_run_to_failed(self):
         self._set_default_task_instance_states(dr)
 
         altered = set_dag_run_state_to_failed(self.dag1, date, commit=True)
-
         # Only running task should be altered.
-        assert len(altered) == 1
+        expected = self._get_num_tasks_with_starting_state(State.RUNNING, inclusion=True)
+        assert len(altered) == expected
         self._verify_dag_run_state(self.dag1, date, State.FAILED)
         assert dr.get_task_instance('run_after_loop').state == State.FAILED
         self._verify_dag_run_dates(self.dag1, date, State.FAILED, middle_time)
@@ -487,7 +498,8 @@ def test_set_success_dag_run_to_success(self):
         altered = set_dag_run_state_to_success(self.dag1, date, commit=True)
 
         # All except the SUCCESS task should be altered.
-        assert len(altered) == 5
+        expected = self._get_num_tasks_with_starting_state(State.SUCCESS, inclusion=False)
+        assert len(altered) == expected
         self._verify_dag_run_state(self.dag1, date, State.SUCCESS)
         self._verify_task_instance_states(self.dag1, date, State.SUCCESS)
         self._verify_dag_run_dates(self.dag1, date, State.SUCCESS, middle_time)
@@ -499,9 +511,9 @@ def test_set_success_dag_run_to_failed(self):
         self._set_default_task_instance_states(dr)
 
         altered = set_dag_run_state_to_failed(self.dag1, date, commit=True)
-
         # Only running task should be altered.
-        assert len(altered) == 1
+        expected = self._get_num_tasks_with_starting_state(State.RUNNING, inclusion=True)
+        assert len(altered) == expected
         self._verify_dag_run_state(self.dag1, date, State.FAILED)
         assert dr.get_task_instance('run_after_loop').state == State.FAILED
         self._verify_dag_run_dates(self.dag1, date, State.FAILED, middle_time)
@@ -529,7 +541,8 @@ def test_set_failed_dag_run_to_success(self):
         altered = set_dag_run_state_to_success(self.dag1, date, commit=True)
 
         # All except the SUCCESS task should be altered.
-        assert len(altered) == 5
+        expected = self._get_num_tasks_with_starting_state(State.SUCCESS, inclusion=False)
+        assert len(altered) == expected
         self._verify_dag_run_state(self.dag1, date, State.SUCCESS)
         self._verify_task_instance_states(self.dag1, date, State.SUCCESS)
         self._verify_dag_run_dates(self.dag1, date, State.SUCCESS, middle_time)
@@ -543,7 +556,8 @@ def test_set_failed_dag_run_to_failed(self):
         altered = set_dag_run_state_to_failed(self.dag1, date, commit=True)
 
         # Only running task should be altered.
-        assert len(altered) == 1
+        expected = self._get_num_tasks_with_starting_state(State.RUNNING, inclusion=True)
+        assert len(altered) == expected
         self._verify_dag_run_state(self.dag1, date, State.FAILED)
         assert dr.get_task_instance('run_after_loop').state == State.FAILED
         self._verify_dag_run_dates(self.dag1, date, State.FAILED, middle_time)
@@ -578,15 +592,17 @@ def test_set_state_without_commit(self):
 
         will_be_altered = set_dag_run_state_to_failed(self.dag1, date, commit=False)
 
-        # Only the running task will be altered.
-        assert len(will_be_altered) == 1
+        # Only the running task shouldbe altered.
+        expected = self._get_num_tasks_with_starting_state(State.RUNNING, inclusion=True)
+        assert len(will_be_altered) == expected
         self._verify_dag_run_state(self.dag1, date, State.RUNNING)
         self._verify_task_instance_states_remain_default(dr)
 
         will_be_altered = set_dag_run_state_to_success(self.dag1, date, commit=False)
 
         # All except the SUCCESS task should be altered.
-        assert len(will_be_altered) == 5
+        expected = self._get_num_tasks_with_starting_state(State.SUCCESS, inclusion=False)
+        assert len(will_be_altered) == expected
         self._verify_dag_run_state(self.dag1, date, State.RUNNING)
         self._verify_task_instance_states_remain_default(dr)
 

tests/cli/commands/test_kubernetes_command.py

@@ -34,13 +34,13 @@ def setUpClass(cls):
 
     def test_generate_dag_yaml(self):
         with tempfile.TemporaryDirectory("airflow_dry_run_test/") as directory:
-            file_name = "example_bash_operator_run_after_loop_2020-11-03T00_00_00_plus_00_00.yml"
+            file_name = "miscellaneous_test_dag_run_after_loop_2020-11-03T00_00_00_plus_00_00.yml"
             kubernetes_command.generate_pod_yaml(
                 self.parser.parse_args(
                     [
                         'kubernetes',
                         'generate-dag-yaml',
-                        'example_bash_operator',
+                        'miscellaneous_test_dag',
                         "2020-11-03",
                         "--output-path",
                         directory,

tests/dags/test_miscellaneous.py

@@ -0,0 +1,75 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Example DAG demonstrating the usage of the BashOperator."""
+
+from datetime import timedelta
+
+from airflow import DAG
+from airflow.operators.bash import BashOperator
+from airflow.operators.dummy import DummyOperator
+from airflow.utils.dates import days_ago
+
+args = {
+    'owner': 'airflow',
+}
+
+dag = DAG(
+    dag_id='miscellaneous_test_dag',
+    default_args=args,
+    schedule_interval='0 0 * * *',
+    start_date=days_ago(2),
+    dagrun_timeout=timedelta(minutes=60),
+    tags=['example', 'example2'],
+    params={"example_key": "example_value"},
+)
+
+run_this_last = DummyOperator(
+    task_id='run_this_last',
+    dag=dag,
+)
+
+# [START howto_operator_bash]
+run_this = BashOperator(
+    task_id='run_after_loop',
+    bash_command='echo 1',
+    dag=dag,
+)
+# [END howto_operator_bash]
+
+run_this >> run_this_last
+
+for i in range(3):
+    task = BashOperator(
+        task_id='runme_' + str(i),
+        bash_command='echo "{{ task_instance_key_str }}" && sleep 1',
+        dag=dag,
+    )
+    task >> run_this
+
+# [START howto_operator_bash_template]
+also_run_this = BashOperator(
+    task_id='also_run_this',
+    bash_command='echo "run_id={{ run_id }} | dag_run={{ dag_run }}"',
+    dag=dag,
+)
+# [END howto_operator_bash_template]
+also_run_this >> run_this_last
+
+if __name__ == "__main__":
+    dag.cli()

tests/jobs/test_backfill_job.py

@@ -165,7 +165,7 @@ def test_trigger_controller_dag(self):
 
     @pytest.mark.backend("postgres", "mysql")
     def test_backfill_multi_dates(self):
-        dag = self.dagbag.get_dag('example_bash_operator')
+        dag = self.dagbag.get_dag('miscellaneous_test_dag')
 
         end_date = DEFAULT_DATE + datetime.timedelta(days=1)
 
@@ -230,7 +230,7 @@ def test_backfill_multi_dates(self):
                 ),
             ],
             [
-                "example_bash_operator",
+                "miscellaneous_test_dag",
                 ("runme_0", "runme_1", "runme_2", "also_run_this", "run_after_loop", "run_this_last"),
             ],
             [