vizeit opened a new issue, #53957: URL: https://github.com/apache/airflow/issues/53957
### Apache Airflow Provider(s) cncf-kubernetes ### Versions of Apache Airflow Providers apache-airflow-providers-cncf-kubernetes==9.0.1 ### Apache Airflow version 2.10.5 ### Operating System Linux ### Deployment Other 3rd-party Helm chart ### Deployment details **Airflow Compute:** GKE - **scheduler replicas:** 2 **Airflow Meta Database:** MySQL 8.0.31 **Main Airflow configuration:** _executor:_ KubernetesExecutor _AIRFLOW__SCHEDULER__SCHEDULE_AFTER_TASK_EXECUTION:_ False _AIRFLOW__CORE__MAX_NUM_RENDERED_TI_FIELDS_PER_TASK:_ 0 _AIRFLOW__CORE__PARALLELISM:_ 250 ### What happened Using @task.kubernetes in a DAG causes dag hash to change every time serialized dag update occurs based on AIRFLOW__CORE__MIN_SERIALIZED_DAG_UPDATE_INTERVAL. The change in dag hash for every serialized dag update only occurs when multiple schedulers are configured. In my case I have 2 schedulers so I see the dag hash flip flop between two different values. After troubleshooting and going through the Airflow 2.10.5 source code, I see the root cause of the issue to be [this code statement](https://github.com/apache/airflow/blob/b93c3db6b1641b0840bd15ac7d05bc58ff2cccbf/airflow/providers/cncf/kubernetes/decorators/kubernetes.py#L60) `template_fields: Sequence[str] = tuple( {"op_args", "op_kwargs", *KubernetesPodOperator.template_fields} - {"cmds", "arguments"} )` template_fields are initialized using set which has non-deterministic order which results into different dag hash on different scheduler. The frequent change in dag hash amplifies update dag_run statements during scheduling `UPDATE `dag_run` SET `last_scheduling_decision` = ?, `dag_hash` = ?, `updated_at` = ? WHERE `dag_run` . `id` = ?` `UPDATE `dag_run` SET `last_scheduling_decision` = ?, `updated_at` = ? WHERE `dag_run` . `id` = ?` `UPDATE `dag_run` SET `dag_hash` = ?, `updated_at` = ? WHERE `dag_run` . `id` = ?` This additional updates causes more lock waits in the database. This especially becomes a huge problem for the DAGs with dynamic task groups expanding into double digit tasks ### What you think should happen instead The order of template_fields should be consistent and should not cause change in dag hash when nothing has changed in the actual DAG ### How to reproduce 1. Create a simple dag with @task.kubernetes ```"""DAG file to load test Dynamic task mapping group""" from airflow.decorators import task, task_group from airflow.models.dag import DAG @task def make_list(no_tasks: str): """Generate and return a list of values Args: no_tasks (str): Number of tasks to generate """ return [i for i in range(int(no_tasks if no_tasks else 100))] @task def consumer(arg): """Consume the list of values""" print(arg) @task def agent(arg): """Agent to handle the task""" print(arg) @task.kubernetes( image="python:3.8-slim-buster", queue="kubernetes", ) def company(arg): """Company to handle the task""" print(arg) @task_group def supplier(arg): """Supply the list of values""" consumer(arg) agent(arg) company(arg) with DAG(dag_id="dynamic-map-group", schedule_interval=None) as dag: no_of_tasks = "{{ dag_run.conf['no_of_tasks'] }}" supplier.expand(arg=make_list(no_of_tasks)) ``` 2. Ensure that multiple schedulers are configured 3. Once DAG is deployed, connect to the meta database and query serialized_dag table `select dag_hash, last_updated, data from serialized_dag where dag_id = 'mydag'` When you query multiple times, you will see the dag_hash changes every configured AIRFLOW__CORE__MIN_SERIALIZED_DAG_UPDATE_INTERVAL 4. Run the dag with a very high number to create large number of tasks in the dynamic task group 5. Observe the database level top queries and lock waits, you will see multiple update statements to dag_run table resulting into higher number of lock waits ### Anything else - I checked the most recent source code for airflow 3 and in my observation the latest version may have the same issue, I do not see template fields being preserved in order during serialization - Similar issue with template_fields order may exist in other places as well that could impact the dag hash and the issue with additional scheduling updates e.g. KubernetesJobOperator - I was able to test with changing the template_fields logic for kubernetes decorator to keep deterministic order and confirm that dag hash does not change and does not result into extra updates to dag_run ### Are you willing to submit PR? - [x] Yes I am willing to submit a PR! ### Code of Conduct - [x] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
