TaskHandlerWithCustomFormatter adds prefix twice

[bellhea]

Apache Airflow version: 1.10.11

Kubernetes version (if you are using kubernetes) (use kubectl version): 1.10.0

Environment:

• Cloud provider or hardware configuration:
• OS (e.g. from /etc/os-release): rhel 7.7
• Kernel (e.g. uname -a):
• Install tools:
• Others:

What happened:
I'm trying to use the class TaskHandlerWithCustomFormatter.
Since I am on version 1.10.11, this class is not yet available so I've copied the class and using it in my custom log_config.py. This is how I've used it:

import logging
from logging import StreamHandler
from airflow.utils.helpers import parse_template_string


class TaskHandlerWithCustomFormatter(StreamHandler):
    """
    Custom implementation of StreamHandler, a class which writes logging records for Airflow
    """
    def __init__(self, stream):
        super().__init__()
        self.prefix_jinja_template = None

    def set_context(self, ti):
        """
        Accept the run-time context (i.e. the current task) and configure the formatter accordingly.
        :param ti:
        :return:
        """
        if ti.raw:
            return
        prefix = "TASK - {{ti.dag_id}}-{{ti.task_id}}-{{ds}}-{{try_number}}"

        rendered_prefix = ""
        if prefix:
            _, self.prefix_jinja_template = parse_template_string(prefix)
            rendered_prefix = self._render_prefix(ti)
        formatter = logging.Formatter(rendered_prefix + ":" + self.formatter._fmt)  # pylint: disable=W0212
        self.setFormatter(formatter)
        self.setLevel(self.level)

    def _render_prefix(self, ti):
        if self.prefix_jinja_template:
            jinja_context = ti.get_template_context()
            return self.prefix_jinja_template.render(**jinja_context)
        logging.warning("'task_log_prefix_template' is in invalid format, ignoring the variable value")
        return ""

'handlers': {
    'console': {
        'class': 'logging.StreamHandler',
        'formatter': 'airflow',
        'stream': 'ext://sys.stdout'
    },
    'task_handler': {
        'class': 'task_handler_with_custom_formatter.TaskHandlerWithCustomFormatter',
        'formatter': 'airflow',
        'stream': 'ext://sys.stdout'
    },
...
'loggers': {
    'airflow.processor': {
        'handlers': ['processor', 'console'],
        'level': LOG_LEVEL,
        'propagate': False,
    },
    'airflow.task': {
        'handlers': ['task_handler'],
        'level': LOG_LEVEL,
        'propagate': False,
    },

I can see the prefix as I have defined in the logs, however the prefix is printed twice before the actual log text, e.g.:

dag_id-task_id-2016-01-01-1:dag_id-task_id-2016-01-01-1:<LOG TEXT>

Any ideas what could be wrong here?

What you expected to happen:
The prefix to only appear once

How to reproduce it:
Code as above

Anything else we need to know: