Override default_args between Nested TaskGroups

[hyeong10]

What do you see as an issue?

Hello!

I don't know if this is intended, but default_args is not an override when using nested TaskGroups.

def callback_in_dag(context: Context):
    print("DAG!")

def callback_in_task_group(context: Context):
    print("Parent TaskGroup!")

with DAG(
    "some_dag_id",
    default_args={
        "on_failure_callback": callback_in_dag
    },
    schedule=None,
    start_date=datetime(2023, 1, 1)
) as dag:
    with TaskGroup(
        "parent_tg", 
        default_args={
            "on_failure_callback": callback_in_task_group
        }
    ) as parent_tg:
        with TaskGroup("child_tg") as child_tg:
            BashOperator(task_id="task_1", bash_command="nooooo_command")

I want the result to be "Parent TaskGroup!", but I get "DAG!".

[2023-05-30, 10:38:52 KST] {logging_mixin.py:137} INFO - DAG!

Solving the problem

Add _update_default_args like link

airflow/utils/task_group.py

...
class TaskGroup(DAGNode):
    def __init__(...):
        ...
        self.default_args = copy.deepcopy(default_args or {})

        # Call 'self._update_default_args' when exists parent_group
        if parent_group is not None:
            self._update_default_args(parent_group)
        ...
    ...
    # Update self.default_args
    def _update_default_args(parent_group: TaskGroup):
        if parent_group.default_args:
            self.default_args.update(parent_group.default_args) 
...

Anything else

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[boring-cyborg]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.