Remove deprecated BashTaskRunner (#12295)

This commit:

- Remove support for BashTaskRunner, this task_runner was deprecated from
Airflow 1.10.3 (https://github.com/apache/airflow/blob/1.10.3/UPDATING.md#rename-of-bashtaskrunner-to-standardtaskrunner)

- Support deprecated `hostname_callable` & `email_backedn` until 2.1 since it has not been deprecated in any relased Airflow versions

airflow/configuration.py

@@ -177,8 +177,7 @@ class AirflowConfigParser(ConfigParser):  # pylint: disable=too-many-ancestors
     # about. Mapping of section -> setting -> { old, replace, by_version }
     deprecated_values = {
         'core': {
-            'task_runner': (re.compile(r'\ABashTaskRunner\Z'), r'StandardTaskRunner', '2.0'),
-            'hostname_callable': (re.compile(r':'), r'.', '2.0'),
+            'hostname_callable': (re.compile(r':'), r'.', '2.1'),
         },
         'webserver': {
             'navbar_color': (re.compile(r'\A#007A87\Z', re.IGNORECASE), '#fff', '2.1'),
@@ -187,7 +186,7 @@ class AirflowConfigParser(ConfigParser):  # pylint: disable=too-many-ancestors
             'email_backend': (
                 re.compile(r'^airflow\.contrib\.utils\.sendgrid\.send_email$'),
                 r'airflow.providers.sendgrid.utils.emailer.send_email',
-                '2.0',
+                '2.1',
             ),
         },
     }

tests/core/test_configuration.py

@@ -489,15 +489,13 @@ def make_config():
             # lookup even if we remove this explicit fallback
             test_conf.deprecated_values = {
                 'core': {
-                    'task_runner': (re.compile(r'\ABashTaskRunner\Z'), r'StandardTaskRunner', '2.0'),
-                    'hostname_callable': (re.compile(r':'), r'.', '2.0'),
+                    'hostname_callable': (re.compile(r':'), r'.', '2.1'),
                 },
             }
             test_conf.read_dict(
                 {
                     'core': {
                         'executor': 'SequentialExecutor',
-                        'task_runner': 'BashTaskRunner',
                         'sql_alchemy_conn': 'sqlite://',
                         'hostname_callable': 'socket:getfqdn',
                     },
@@ -507,33 +505,21 @@ def make_config():
 
         with self.assertWarns(FutureWarning):
             test_conf = make_config()
-            self.assertEqual(test_conf.get('core', 'task_runner'), 'StandardTaskRunner')
             self.assertEqual(test_conf.get('core', 'hostname_callable'), 'socket.getfqdn')
 
-        with self.assertWarns(FutureWarning):
-            with unittest.mock.patch.dict('os.environ', AIRFLOW__CORE__TASK_RUNNER='BashTaskRunner'):
-                test_conf = make_config()
-
-                self.assertEqual(test_conf.get('core', 'task_runner'), 'StandardTaskRunner')
-
         with self.assertWarns(FutureWarning):
             with unittest.mock.patch.dict('os.environ', AIRFLOW__CORE__HOSTNAME_CALLABLE='socket:getfqdn'):
                 test_conf = make_config()
-
                 self.assertEqual(test_conf.get('core', 'hostname_callable'), 'socket.getfqdn')
 
         with reset_warning_registry():
             with warnings.catch_warnings(record=True) as warning:
                 with unittest.mock.patch.dict(
                     'os.environ',
-                    AIRFLOW__CORE__TASK_RUNNER='NotBashTaskRunner',
                     AIRFLOW__CORE__HOSTNAME_CALLABLE='CarrierPigeon',
                 ):
                     test_conf = make_config()
-
-                    self.assertEqual(test_conf.get('core', 'task_runner'), 'NotBashTaskRunner')
                     self.assertEqual(test_conf.get('core', 'hostname_callable'), 'CarrierPigeon')
-
                     self.assertListEqual([], warning)
 
     def test_deprecated_funcs(self):