**Apache Airflow version**: 2.1.0


**Environment**: tested on Windows docker-compose envirnoment and on k8s (both with celery executor).

**What happened**:

```
[2021-06-16 07:56:32,682] {taskinstance.py:1481} ERROR - Task failed with exception
Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1137, in _run_raw_task
    self._prepare_and_execute_task_with_callbacks(context, task)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1311, in _prepare_and_execute_task_with_callbacks
    result = self._execute_task(context, task_copy)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1341, in _execute_task
    result = task_copy.execute(context=context)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/operators/trigger_dagrun.py", line 134, in execute
    replace_microseconds=False,
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/api/common/experimental/trigger_dag.py", line 123, in trigger_dag
    replace_microseconds=replace_microseconds,
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/api/common/experimental/trigger_dag.py", line 48, in _trigger_dag
    dag = dag_bag.get_dag(dag_id)  # prefetch dag if it is stored serialized
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/utils/session.py", line 70, in wrapper
    return func(*args, session=session, **kwargs)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/models/dagbag.py", line 186, in get_dag
    self._add_dag_from_db(dag_id=dag_id, session=session)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/models/dagbag.py", line 252, in _add_dag_from_db
    dag = row.dag
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/models/serialized_dag.py", line 175, in dag
    dag = SerializedDAG.from_dict(self.data)  # type: Any
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/serialization/serialized_objects.py", line 792, in from_dict
    return cls.deserialize_dag(serialized_obj['dag'])
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/serialization/serialized_objects.py", line 716, in deserialize_dag
    v = {task["task_id"]: SerializedBaseOperator.deserialize_operator(task) for task in v}
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/serialization/serialized_objects.py", line 716, in <dictcomp>
    v = {task["task_id"]: SerializedBaseOperator.deserialize_operator(task) for task in v}
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/serialization/serialized_objects.py", line 493, in deserialize_operator
    op_predefined_extra_links = cls._deserialize_operator_extra_links(v)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/serialization/serialized_objects.py", line 600, in _deserialize_operator_extra_links
    if _operator_link_class_path in get_operator_extra_links():
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/serialization/serialized_objects.py", line 86, in get_operator_extra_links
    _OPERATOR_EXTRA_LINKS.update(ProvidersManager().extra_links_class_names)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/providers_manager.py", line 400, in extra_links_class_names
    self.initialize_providers_manager()
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/providers_manager.py", line 129, in initialize_providers_manager
    self._discover_all_providers_from_packages()
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/providers_manager.py", line 151, in _discover_all_providers_from_packages
    log.debug("Loading %s from package %s", entry_point, package_name)
  File "/usr/local/lib/python3.7/logging/__init__.py", line 1366, in debug
    self._log(DEBUG, msg, args, **kwargs)
  File "/usr/local/lib/python3.7/logging/__init__.py", line 1514, in _log
    self.handle(record)
  File "/usr/local/lib/python3.7/logging/__init__.py", line 1524, in handle
    self.callHandlers(record)
  File "/usr/local/lib/python3.7/logging/__init__.py", line 1586, in callHandlers
    hdlr.handle(record)
  File "/usr/local/lib/python3.7/logging/__init__.py", line 890, in handle
    rv = self.filter(record)
  File "/usr/local/lib/python3.7/logging/__init__.py", line 751, in filter
    result = f.filter(record)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/utils/log/secrets_masker.py", line 157, in filter
    record.__dict__[k] = self.redact(v)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/utils/log/secrets_masker.py", line 203, in redact
    return tuple(self.redact(subval) for subval in item)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/utils/log/secrets_masker.py", line 203, in <genexpr>
    return tuple(self.redact(subval) for subval in item)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/utils/log/secrets_masker.py", line 203, in redact
    return tuple(self.redact(subval) for subval in item)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/utils/log/secrets_masker.py", line 203, in <genexpr>
    return tuple(self.redact(subval) for subval in item)
  ....
    return tuple(self.redact(subval) for subval in item)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/utils/log/secrets_masker.py", line 203, in redact
    return tuple(self.redact(subval) for subval in item)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/utils/log/secrets_masker.py", line 203, in <genexpr>
    return tuple(self.redact(subval) for subval in item)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/utils/log/secrets_masker.py", line 203, in redact
    return tuple(self.redact(subval) for subval in item)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/utils/log/secrets_masker.py", line 203, in <genexpr>
    return tuple(self.redact(subval) for subval in item)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/utils/log/secrets_masker.py", line 201, in redact
    elif isinstance(item, (tuple, set)):
RecursionError: maximum recursion depth exceeded in __instancecheck__ 
```

**What you expected to happen**:
I think new masker is not able to handle TriggerDagRunOperator running dag with TriggerDagRunOperator

**How to reproduce it**:
```
from datetime import datetime, timedelta

from airflow.models import DAG
from airflow.operators.python_operator import PythonOperator
from airflow.operators.trigger_dagrun import TriggerDagRunOperator

def pprint(**kwargs):
    print(1)

with DAG("test",
         catchup=False,
         max_active_runs=1,
         start_date=datetime(2021, 1, 1),
         is_paused_upon_creation=False,
         schedule_interval=None) as dag:

    task_observe_pea_data = PythonOperator(
        task_id="test_task",
        python_callable=pprint,
        provide_context=True
    )

with DAG("test_1",
         catchup=False,
         max_active_runs=1,
         start_date=datetime(2021, 1, 1),
         is_paused_upon_creation=False,
         schedule_interval=None) as dag:

    task_observe_pea_data = TriggerDagRunOperator(
        task_id="test_trigger_1",
        trigger_dag_id="test"
    )

with DAG("test_2",
         catchup=False,
         max_active_runs=1,
         start_date=datetime(2021, 1, 1),
         is_paused_upon_creation=False,
         schedule_interval=None) as dag:

    task_observe_pea_data = TriggerDagRunOperator(
        task_id="test_trigger_2",
        trigger_dag_id="test_1"
    )
```

**Anything else we need to know**:

How often does this problem occur? Every time
I have tried hide_sensitive_var_conn_fields=False but error still occurs.
airflow/utils/log/secrets_masker.py
#
#
"""Mask sensitive information from logs"""
import collections
import logging
import re
from typing import TYPE_CHECKING, Iterable, Optional, Set, TypeVar, Union
from airflow.compat.functools import cache, cached_property
if TYPE_CHECKING:
    from airflow.typing_compat import RePatternType
    RedactableItem = TypeVar('RedactableItem')
log = logging.getLogger(__name__)
DEFAULT_SENSITIVE_FIELDS = frozenset(
    {
        'access_token',
        'api_key',
        'apikey',
        'authorization',
        'passphrase',
        'passwd',
        'password',
        'private_key',
        'secret',
    }
)
"""Names of fields (Connection extra, Variable key name etc.) that are deemed sensitive"""
@cache
def get_sensitive_variables_fields():