retries from default_args are not respected

[ktaborski]

Apache Airflow version

3.1.1

If "Other Airflow 2/3 version" selected, which one?

No response

What happened?

In my airflow.cfg I have:

[core]
...
default_task_retries = 3
...

This is example DAG:

from datetime import datetime
from airflow.sdk import DAG
from airflow.providers.standard.operators.bash import BashOperator

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2024, 1, 1),
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 0,
}

dag = DAG(
    'test_dag',
    default_args=default_args,
    description='Test DAG that echoes a message and fails',
    schedule=None,
    catchup=False,
    tags=['test'],
)

test_task = BashOperator(
    task_id='echo_and_fail',
    bash_command='echo "This is a test message" && exit 1',
    dag=dag,
)

test_task is executed 4 times (1 + 3 retries), although in default_args I have 'retries': 0.

In version 3.0.6 it is working as expected

What you think should happen instead?

Task should be executed only once.

How to reproduce

Steps provided in description

Operating System

EKS bottlerocket

Versions of Apache Airflow Providers

No response

Deployment

Official Apache Airflow Helm Chart

Deployment details

No response

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

[tirkarthi]

This could be due to d9969be where schema has 0 as default value for "retries" which could be leading to it not being serialized.

airflow/airflow-core/src/airflow/serialization/serialized_objects.py

Lines 1868 to 1869 in 1242e07

	if schema_defaults[attrname] == var:
	return True

{"retries": 1} in default_args has the retries value in tasks dict

{"dag": {"tags": ["test"], "tasks": [{"__var": {"retries": 1, "task_id": "echo_and_fail", "ui_color": "#f0ede4", "task_type": "BashOperator", "retry_delay": 300.0, "_task_module": "airflow.providers.standard.operators.bash", "bash_command": "echo \"This is a test message\" && exit 1", "template_ext": [".sh", ".bash"], "template_fields": ["bash_command", "env", "cwd"], "_needs_expansion": false, "template_fields_renderers": {"env": "json", "bash_command": "bash"}}, "__type": "operator"}], "dag_id": "gh57792", "params": [], "fileloc": "/home/karthikeyan/airflow/dags1/gh57792.py", "deadline": null, "timezone": "UTC", "edge_info": {}, "timetable": {"__var": {}, "__type": "airflow.timetables.simple.NullTimetable"}, "start_date": 1704067200.0, "task_group": {"tooltip": "", "children": {"echo_and_fail": ["operator", "echo_and_fail"]}, "ui_color": "CornflowerBlue", "_group_id": null, "ui_fgcolor": "#000", "prefix_group_id": true, "upstream_task_ids": [], "group_display_name": "", "upstream_group_ids": [], "downstream_task_ids": [], "downstream_group_ids": []}, "description": "Test DAG that echoes a message and fails", "default_args": {"__var": {"owner": "airflow", "retries": 1, "start_date": {"__var": 1704067200.0, "__type": "datetime"}, "email_on_retry": false, "depends_on_past": false, "email_on_failure": false}, "__type": "dict"}, "dag_dependencies": [], "relative_fileloc": "gh57792.py", "_processor_dags_folder": "/home/karthikeyan/airflow/dags"}, "__version": 3, "client_defaults": {"tasks": {"retries": 10}}}

{"retries": 0} in default_args skips the retries value in tasks dict

{"dag": {"tags": ["test"], "tasks": [{"__var": {"task_id": "echo_and_fail", "ui_color": "#f0ede4", "task_type": "BashOperator", "retry_delay": 300.0, "_task_module": "airflow.providers.standard.operators.bash", "bash_command": "echo \"This is a test message\" && exit 1", "template_ext": [".sh", ".bash"], "template_fields": ["bash_command", "env", "cwd"], "_needs_expansion": false, "template_fields_renderers": {"env": "json", "bash_command": "bash"}}, "__type": "operator"}], "dag_id": "gh57792", "params": [], "fileloc": "/home/karthikeyan/airflow/dags1/gh57792.py", "deadline": null, "timezone": "UTC", "edge_info": {}, "timetable": {"__var": {}, "__type": "airflow.timetables.simple.NullTimetable"}, "start_date": 1704067200.0, "task_group": {"tooltip": "", "children": {"echo_and_fail": ["operator", "echo_and_fail"]}, "ui_color": "CornflowerBlue", "_group_id": null, "ui_fgcolor": "#000", "prefix_group_id": true, "upstream_task_ids": [], "group_display_name": "", "upstream_group_ids": [], "downstream_task_ids": [], "downstream_group_ids": []}, "description": "Test DAG that echoes a message and fails", "default_args": {"__var": {"owner": "airflow", "retries": 0, "start_date": {"__var": 1704067200.0, "__type": "datetime"}, "email_on_retry": false, "depends_on_past": false, "email_on_failure": false}, "__type": "dict"}, "dag_dependencies": [], "relative_fileloc": "gh57792.py", "_processor_dags_folder": "/home/karthikeyan/airflow/dags"}, "__version": 3, "client_defaults": {"tasks": {"retries": 10}}}

[amoghrajesh]

Thanks @tirkarthi, good investigation.

I was able to achieve something similar in terms on investigation and was suspecting similar issue too:

os.environ['AIRFLOW__CORE__DEFAULT_TASK_RETRIES'] = '3'

def return_deserialized_task_after_serializng(retries):
    from datetime import datetime
    from airflow.sdk import DAG
    from airflow.providers.standard.operators.bash import BashOperator
    from airflow.serialization.serialized_objects import SerializedDAG
    from airflow.sdk.bases.operator import OPERATOR_DEFAULTS
    dag_0 = DAG(f'test_{retries}', default_args={'retries': retries}, start_date=datetime(2024, 1, 1))
    task_0 = BashOperator(task_id=f't{retries}', bash_command='echo', dag=dag_0)
    serialized_0 = SerializedDAG.to_dict(dag_0)
    task_data_0 = serialized_0["dag"]["tasks"][0]["__var"]
    client_defaults = serialized_0.get("client_defaults", {}).get("tasks", {})
    deserialized_dag_0 = SerializedDAG.from_dict(serialized_0)
    deserialized_task_0 = deserialized_dag_0.get_task(f't{retries}')
    
    return deserialized_task_0

When I call this with retries = 0 and 1, I get:

ds_task_0 = return_deserialized_task_after_serializng(0)
ds_task_0.retries
Out[8]: 3

ds_task_1 = return_deserialized_task_after_serializng(1)
ds_task_1.retries
Out[16]: 1

It is overriden by conf at runtime. Let me try a fix

[matthieuauger]

Hello, is it possible that this feature broke the @retries settings at task level?

I have a retries: 3 in default_args on my DAG, but also a @task(retries=0) on a task. Since a few days this tasks sometimes get retried multiple times.

I'm going to open a proper issue on this but wanted to share it here because it may be due to this fix