Fix apache#35622: TaskHandlerWithCustomFormatter now adds prefix only…

… once

When using the TaskHandlerWithCustomFormatter to add a prefix to logs, it was previously adding the prefix multiple times. This happened because it was being called multiple times from logging_mixin.py, and worsened because even when the handler's formatter was a TimezoneAware formatter (to include UTC offset), it was still adding an additional prefix. Because of this, I felt that any solution outside of the TaskHandlerWithCustomFormatter itself would either require a restructuring of the handlers' structure or slow down execution for all other handlers. And so, the solution I settled on was to add to TaskHandlerWithCustomFormatter's initial 'if' statement a simple 'or self.prefix_jinja_template is not None', so that it returns if the prefix had already been set. This is similar to what is done by the ElasticSearch es_task_handler.py.

Note: also fixed the documentation's example for the handler, as the previous one was incorrect and didn't work.

airflow/config_templates/config.yml

@@ -848,7 +848,7 @@ logging:
         Specify prefix pattern like mentioned below with stream handler TaskHandlerWithCustomFormatter
       version_added: 2.0.0
       type: string
-      example: "{ti.dag_id}-{ti.task_id}-{execution_date}-{try_number}"
+      example: "{{ti.dag_id}}-{{ti.task_id}}-{{execution_date}}-{{ti.try_number}}"
       is_template: true
       default: ""
     log_filename_template:

airflow/utils/log/task_handler_with_custom_formatter.py

@@ -45,7 +45,7 @@ def set_context(self, ti) -> None:
         :param ti:
         :return:
         """
-        if ti.raw or self.formatter is None:
+        if ti.raw or self.formatter is None or self.prefix_jinja_template is not None:
             return
         prefix = conf.get("logging", "task_log_prefix_template")