Add default operators callbacks to airflow configs

airflow/config_templates/config.yml

@@ -946,6 +946,38 @@
       type: string
       example: ~
       default: "False"
+    - name: default_on_failure_callback
+      description: |
+        The default on_failure_callback assigned to each new operator, unless
+        provided explicitly or passed via ``default_args``
+      version_added: 2.3.0
+      type: string
+      example: ~
+      default: ~
+    - name: default_on_execute_callback
+      description: |
+        The default default_on_execute_callback assigned to each new operator, unless
+        provided explicitly or passed via ``default_args``
+      version_added: 2.3.0
+      type: string
+      example: ~
+      default: ~
+    - name: default_on_retry_callback
+      description: |
+        The default default_on_retry_callback assigned to each new operator, unless
+        provided explicitly or passed via ``default_args``
+      version_added: 2.3.0
+      type: string
+      example: ~
+      default: ~
+    - name: default_on_success_callback
+      description: |
+        The default default_on_success_callback assigned to each new operator, unless
+        provided explicitly or passed via ``default_args``
+      version_added: 2.3.0
+      type: string
+      example: ~
+      default: ~
 - name: hive
   description: ~
   options:

airflow/config_templates/default_airflow.cfg

@@ -491,6 +491,22 @@ default_queue = default
 # If set to False, an exception will be thrown, otherwise only the console message will be displayed.
 allow_illegal_arguments = False
 
+# The default on_failure_callback assigned to each new operator, unless
+# provided explicitly or passed via ``default_args``
+# default_on_failure_callback =
+
+# The default default_on_execute_callback assigned to each new operator, unless
+# provided explicitly or passed via ``default_args``
+# default_on_execute_callback =
+
+# The default default_on_retry_callback assigned to each new operator, unless
+# provided explicitly or passed via ``default_args``
+# default_on_retry_callback =
+
+# The default default_on_success_callback assigned to each new operator, unless
+# provided explicitly or passed via ``default_args``
+# default_on_success_callback =
+
 [hive]
 # Default mapreduce queue for HiveOperator tasks
 default_hive_mapred_queue =

airflow/models/baseoperator.py

@@ -550,10 +550,18 @@ def __init__(
         pool_slots: int = 1,
         sla: Optional[timedelta] = None,
         execution_timeout: Optional[timedelta] = None,
-        on_execute_callback: Optional[TaskStateChangeCallback] = None,
-        on_failure_callback: Optional[TaskStateChangeCallback] = None,
-        on_success_callback: Optional[TaskStateChangeCallback] = None,
-        on_retry_callback: Optional[TaskStateChangeCallback] = None,
+        on_execute_callback: Optional[TaskStateChangeCallback] = conf.getimport(
+            'operators', 'default_on_execute_callback', fallback=None
+        ),
+        on_failure_callback: Optional[TaskStateChangeCallback] = conf.getimport(
+            'operators', 'default_on_failure_callback', fallback=None
+        ),
+        on_success_callback: Optional[TaskStateChangeCallback] = conf.getimport(
+            'operators', 'default_on_success_callback', fallback=None
+        ),
+        on_retry_callback: Optional[TaskStateChangeCallback] = conf.getimport(
+            'operators', 'default_on_retry_callback', fallback=None
+        ),
         pre_execute: Optional[TaskPreExecuteHook] = None,
         post_execute: Optional[TaskPostExecuteHook] = None,
         trigger_rule: str = DEFAULT_TRIGGER_RULE,