1fanwang opened a new issue, #54381: URL: https://github.com/apache/airflow/issues/54381
### Apache Airflow version Other Airflow 2 version (please specify below) ### If "Other Airflow 2 version" selected, which one? 2.9.2.32 ### What happened? Kubernetes worker pods are not being automatically deleted after task completion, despite having both DELETE_WORKER_PODS and DELETE_WORKER_PODS_ON_FAILURE set to True. The pods remain in "Error" status and continue consuming cluster resources. Key observations: Task execution completes successfully (marked as SUCCESS in Airflow) Pod shows "Error" status with exit code 1 Pod is not deleted automatically and persists indefinitely Database connection timeout occurs during job cleanup phase Task Execution Flow: Task starts and runs successfully (logs show "test" output and task marked as SUCCESS) Task completes at 2025-08-12T02:17:43.445+0000 Database timeout occurs during job cleanup: Max connect timeout reached while reaching hostgroup 2112221 after 10080ms Pod exits with code 1 due to the database error Pod remains in cluster despite deletion configuration From a quick scan through the [2.9.2 k8s executor src code](https://github.com/apache/airflow/blob/2.9.2/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py), looks like # 1. Task completes successfully [2025-08-12T02:17:14.631+0000] INFO - test # 2. Job cleanup tries to update database Job.complete_execution() → Database timeout # 3. sync() finds empty result_queue def sync(self): with contextlib.suppress(Empty): while True: results = self.result_queue.get_nowait() self._change_state(...) # ← Never called # 4. Pod deletion never happens def _change_state(self, key, state, pod_name, namespace): if self.kube_config.delete_worker_pods: self.kube_scheduler.delete_pod(...) # ← Never deleted ### What you think should happen instead? Worker pods should be automatically deleted after task completion, regardless of whether the cleanup phase encounters database connectivity issues. The pod deletion mechanism should be resilient to exceptions / database timeouts that occur after successful task execution, or at least retry. ### How to reproduce Configure Kubernetes executor with the following environment variables: yaml- name: AIRFLOW__KUBERNETES__DELETE_WORKER_PODS value: 'True' - name: AIRFLOW__KUBERNETES__DELETE_WORKER_PODS_ON_FAILURE value: 'True' Run a DAG with tasks that may experience database connectivity issues during cleanup Observe that pods remain in "Error" status and are not deleted ### Operating System NAME="Common Base Linux Mariner" VERSION="2.0.20241006" ID=mariner VERSION_ID="2.0" PRETTY_NAME="CBL-Mariner/Linux" ### Versions of Apache Airflow Providers airflow [ /opt/airflow ]$ /opt/airflow/airflow2.9/bin/pip list | grep provider ``` apache-airflow-providers-cncf-kubernetes 8.4.2 apache-airflow-providers-common-compat 1.6.1 apache-airflow-providers-common-io 1.5.4 apache-airflow-providers-common-sql 1.26.0 apache-airflow-providers-fab 1.5.3 apache-airflow-providers-ftp 3.12.2 apache-airflow-providers-http 5.2.0 apache-airflow-providers-imap 3.8.2 apache-airflow-providers-lnkd 9.0.32 apache-airflow-providers-smtp 2.0.3 apache-airflow-providers-sqlite 3.3.1 ``` ### Deployment Official Apache Airflow Helm Chart ### Deployment details Airflow running on Kubernetes with KubernetesExecutor Using persistent volumes for DAGs and logs Database: MySQL with connection pooling Load testing environment with high task concurrency Configuration: AIRFLOW__KUBERNETES__DELETE_WORKER_PODS: 'True' AIRFLOW__KUBERNETES__DELETE_WORKER_PODS_ON_FAILURE: 'True' AIRFLOW__CORE__PARALLELISM: 1024 AIRFLOW__CORE__MAX_ACTIVE_TASKS_PER_DAG: 2000 Error Details: sqlalchemy.exc.OperationalError: (MySQLdb.OperationalError) (9001, 'Max connect timeout reached while reaching hostgroup 2112221 after 10080ms') Pod Status: NAME: workflow-1754956723-0007-dag-7-shape-no-structure-task-029w5t68 STATUS: Failed (0/1 Error, Exit Code: 1) REASON: Error ### Anything else? Task Pod Log ``` [2025-08-12T02:17:43.445+0000] {taskinstance.py:1207} INFO - Marking task as SUCCESS. dag_id=workflow__1754956723_0007__dag_7__shape_no_structure__tasks_1000, task_id=tasks__248_of_1000, run_id=scheduled__2025-08-12T01:15:00+00:00, execution_date=20250812T011500, start_date=20250812T015711, end_date=20250812T021743 [2025-08-12T02:17:43.488+0000] {task_listener.py:103} ERROR - Failed to emit on_task_instance_success event! [2025-08-12T02:17:43.597+0000] {local_task_job_runner.py:240} INFO - Task exited with return code 0 [2025-08-12T02:17:43.597+0000] {local_task_job_runner.py:222} INFO - ::endgroup:: Traceback (most recent call last): File "/opt/airflow/airflow2.9/lib/python3.10/site-packages/sqlalchemy/engine/base.py", line 1910, in _execute_context self.dialect.do_execute( File "/opt/airflow/airflow2.9/lib/python3.10/site-packages/sqlalchemy/engine/default.py", line 736, in do_execute cursor.execute(statement, parameters) File "/opt/airflow/airflow2.9/lib/python3.10/site-packages/MySQLdb/cursors.py", line 179, in execute res = self._query(mogrified_query) File "/opt/airflow/airflow2.9/lib/python3.10/site-packages/MySQLdb/cursors.py", line 330, in _query db.query(q) File "/opt/airflow/airflow2.9/lib/python3.10/site-packages/MySQLdb/connections.py", line 261, in query _mysql.connection.query(self, query) MySQLdb.OperationalError: (9001, 'Max connect timeout reached while reaching hostgroup 2112221 after 10080ms') The above exception was the direct cause of the following exception: Traceback (most recent call last): File "/opt/airflow/airflow2.9/bin/airflow", line 7, in <module> sys.exit(main()) File "/opt/airflow/airflow2.9/lib/python3.10/site-packages/airflow/__main__.py", line 58, in main args.func(args) File "/opt/airflow/airflow2.9/lib/python3.10/site-packages/airflow/cli/cli_config.py", line 49, in command return func(*args, **kwargs) File "/opt/airflow/airflow2.9/lib/python3.10/site-packages/airflow/utils/cli.py", line 114, in wrapper return f(*args, **kwargs) File "/opt/airflow/airflow2.9/lib/python3.10/site-packages/airflow/cli/commands/task_command.py", line 441, in task_run task_return_code = _run_task_by_selected_method(args, _dag, ti) File "/opt/airflow/airflow2.9/lib/python3.10/site-packages/airflow/cli/commands/task_command.py", line 219, in _run_task_by_selected_method return _run_task_by_local_task_job(args, ti) File "/opt/airflow/airflow2.9/lib/python3.10/site-packages/airflow/cli/commands/task_command.py", line 281, in _run_task_by_local_task_job ret = run_job(job=job_runner.job, execute_callable=job_runner._execute) File "/opt/airflow/airflow2.9/lib/python3.10/site-packages/airflow/utils/session.py", line 79, in wrapper return func(*args, session=session, **kwargs) File "/opt/airflow/airflow2.9/lib/python3.10/site-packages/airflow/jobs/job.py", line 417, in run_job job.complete_execution(session=session) File "/opt/airflow/airflow2.9/lib/python3.10/site-packages/airflow/utils/session.py", line 76, in wrapper return func(*args, **kwargs) File "/opt/airflow/airflow2.9/lib/python3.10/site-packages/airflow/jobs/job.py", line 245, in complete_execution Job._update_in_db(job=self, session=session) File "/opt/airflow/airflow2.9/lib/python3.10/site-packages/airflow/api_internal/internal_api_call.py", line 115, in wrapper return func(*args, **kwargs) File "/opt/airflow/airflow2.9/lib/python3.10/site-packages/airflow/utils/session.py", line 76, in wrapper return func(*args, **kwargs) File "/opt/airflow/airflow2.9/lib/python3.10/site-packages/airflow/jobs/job.py", line 353, in _update_in_db session.merge(job) File "/opt/airflow/airflow2.9/lib/python3.10/site-packages/sqlalchemy/orm/session.py", line 3056, in merge return self._merge( File "/opt/airflow/airflow2.9/lib/python3.10/site-packages/sqlalchemy/orm/session.py", line 3136, in _merge merged = self.get( File "/opt/airflow/airflow2.9/lib/python3.10/site-packages/sqlalchemy/orm/session.py", line 2853, in get return self._get_impl( File "/opt/airflow/airflow2.9/lib/python3.10/site-packages/sqlalchemy/orm/session.py", line 2975, in _get_impl return db_load_fn( File "/opt/airflow/airflow2.9/lib/python3.10/site-packages/sqlalchemy/orm/loading.py", line 530, in load_on_pk_identity session.execute( File "/opt/airflow/airflow2.9/lib/python3.10/site-packages/sqlalchemy/orm/session.py", line 1717, in execute result = conn._execute_20(statement, params or {}, execution_options) File "/opt/airflow/airflow2.9/lib/python3.10/site-packages/sqlalchemy/engine/base.py", line 1710, in _execute_20 return meth(self, args_10style, kwargs_10style, execution_options) File "/opt/airflow/airflow2.9/lib/python3.10/site-packages/sqlalchemy/sql/elements.py", line 334, in _execute_on_connection return connection._execute_clauseelement( File "/opt/airflow/airflow2.9/lib/python3.10/site-packages/sqlalchemy/engine/base.py", line 1577, in _execute_clauseelement ret = self._execute_context( File "/opt/airflow/airflow2.9/lib/python3.10/site-packages/sqlalchemy/engine/base.py", line 1953, in _execute_context self._handle_dbapi_exception( File "/opt/airflow/airflow2.9/lib/python3.10/site-packages/sqlalchemy/engine/base.py", line 2134, in _handle_dbapi_exception util.raise_( File "/opt/airflow/airflow2.9/lib/python3.10/site-packages/sqlalchemy/util/compat.py", line 211, in raise_ raise exception File "/opt/airflow/airflow2.9/lib/python3.10/site-packages/sqlalchemy/engine/base.py", line 1910, in _execute_context self.dialect.do_execute( File "/opt/airflow/airflow2.9/lib/python3.10/site-packages/sqlalchemy/engine/default.py", line 736, in do_execute cursor.execute(statement, parameters) File "/opt/airflow/airflow2.9/lib/python3.10/site-packages/MySQLdb/cursors.py", line 179, in execute res = self._query(mogrified_query) File "/opt/airflow/airflow2.9/lib/python3.10/site-packages/MySQLdb/cursors.py", line 330, in _query db.query(q) File "/opt/airflow/airflow2.9/lib/python3.10/site-packages/MySQLdb/connections.py", line 261, in query _mysql.connection.query(self, query) sqlalchemy.exc.OperationalError: (MySQLdb.OperationalError) (9001, 'Max connect timeout reached while reaching hostgroup 2112221 after 10080ms') [SQL: SELECT job.id AS job_id, job.dag_id AS job_dag_id, job.state AS job_state, job.job_type AS job_job_type, job.start_date AS job_start_date, job.end_date AS job_end_date, job.latest_heartbeat AS job_latest_heartbeat, job.executor_class AS job_executor_class, job.hostname AS job_hostname, job.unixname AS job_unixname FROM job WHERE job.id = %s] [parameters: (2409757,)] (Background on this error at: https://sqlalche.me/e/14/e3q8) ``` Task Pod ``` (base) ➜ airflow git:(stewang/enable-worker-pod-infra-failure-cleanup) ✗ kd pod workflow-1754956723-0007-dag-7-shape-no-structure-task-029w5t68 Name: workflow-1754956723-0007-dag-7-shape-no-structure-task-029w5t68 Namespace: oklahoma Priority: 0 Service Account: airflow-load-test-worker Node: <NODE> Start Time: Mon, 11 Aug 2025 18:56:55 -0700 Labels: airflow-worker=2355153 airflow_version=2.9.2.32 component=worker dag_id=workflow__1754956723_0007__dag_7__shape_no_structure_-978241467 kubernetes_executor=True release=airflow-load-test run_id=scheduled__2025-08-12T0115000000-449e16799 task_id=tasks__248_of_1000 tier=airflow try_number=1 Annotations: dag_id: workflow__1754956723_0007__dag_7__shape_no_structure__tasks_1000 run_id: scheduled__2025-08-12T01:15:00+00:00 task_id: tasks__248_of_1000 try_number: 1 Status: Failed IP: 100.97.206.34 IPs: IP: 100.97.206.34 Init Containers: k8s-lare: Container ID: containerd://f8000bae268e2bf083ffd44e3675c2a17be0f455e8be4f76cd6f78eacb9ddd31 Image: <IMAGE> Image ID: <IMAGEID> Port: <none> Host Port: <none> Command: /bin/sh -c Args: sudo -E /k8s-lare request-app-cert appinfo --application ${SELF_MP_NAME} --fabric grid1 --cert-dir /var/cluster/mp --cert-file-owner 50000 --cert-file-group 0; sudo -E /k8s-lare request-app-cert appinfo --application ${SELF_DAG_ID} --fabric grid1 --cert-dir /var/cluster --cert-file-owner 50000 --cert-file-group 0; sudo -E /k8s-lare request-app-cert appinfo --application airflow-main-airflow-oklahoma --fabric grid1 --cert-dir /var/cluster/oklahoma --cert-file-owner 50000 --cert-file-group 0 State: Terminated Reason: Completed Exit Code: 0 Started: Mon, 11 Aug 2025 18:56:56 -0700 Finished: Mon, 11 Aug 2025 18:56:57 -0700 Ready: True Restart Count: 0 Limits: cpu: 1 memory: 1Gi Requests: cpu: 500m memory: 500Mi Environment: SELF_POD_NAME: workflow-1754956723-0007-dag-7-shape-no-structure-task-029w5t68 (v1:metadata.name) SELF_POD_NAMESPACE: oklahoma (v1:metadata.namespace) SELF_DAG_ID: (v1:metadata.annotations['okl_dag_id']) SELF_MP_NAME: (v1:metadata.annotations['okl_mp_name']) Mounts: /var/cluster from dag-certs (rw) /var/cluster/mp from mp-certs (rw) /var/cluster/oklahoma from airflow-main-airflow-oklahoma-app-certs (rw) /var/run/lipki/pki-agent.sock from pki-agent-socket (rw) /var/run/secrets/kubernetes.io/serviceaccount from kube-api-access-jfss5 (ro) Containers: base: Container ID: containerd://b8aaa6fee78dfc514db05a4bcba37af30f4a61adf10667850c202c7d0f16a589 Image: <IMAGE> Image ID: <IMAGEID> Port: <none> Host Port: <none> Args: /opt/airflow/airflow2.9/bin/airflow tasks run workflow__1754956723_0007__dag_7__shape_no_structure__tasks_1000 tasks__248_of_1000 scheduled__2025-08-12T01:15:00+00:00 --local --subdir DAGS_FOLDER/1754956723/workflow__1754956723_0007__dag_7__shape_no_structure__tasks_1000.py State: Terminated Reason: Error Exit Code: 1 Started: Mon, 11 Aug 2025 18:56:57 -0700 Finished: Mon, 11 Aug 2025 19:17:54 -0700 Ready: False Restart Count: 0 Limits: cpu: 500m memory: 1Gi Requests: cpu: 200m memory: 500Mi Environment Variables from: oklahoma-airflow-webserver-sso-client-credential Secret Optional: false Environment: AIRFLOW__CORE__EXECUTOR: LocalExecutor AIRFLOW__CORE__FERNET_KEY: <set to the key 'fernet-key' in secret 'airflow-load-test-fernet-key'> Optional: false AIRFLOW_HOME: /opt/airflow AIRFLOW__CORE__SQL_ALCHEMY_CONN: <set to the key 'connection' in secret 'airflow-load-test-database-secret'> Optional: false AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: <set to the key 'connection' in secret 'airflow-load-test-database-secret'> Optional: false AIRFLOW_CONN_AIRFLOW_DB: <set to the key 'connection' in secret 'airflow-load-test-database-secret'> Optional: false AIRFLOW__WEBSERVER__SECRET_KEY: <set to the key 'webserver-secret-key' in secret 'airflow-load-test-webserver-secret-key'> Optional: false POD_NAME: workflow-1754956723-0007-dag-7-shape-no-structure-task-029w5t68 (v1:metadata.name) POD_NAMESPACE: oklahoma (v1:metadata.namespace) HOSTNAME: (v1:spec.nodeName) HOST_IP: (v1:status.hostIP) AIRFLOW__METRICS__OTEL_HOST: ${HOST_IP} AIRFLOW__METRICS__OTEL_RESOURCE_ATTRIBUTES: { "_microsoft_metrics_account": "LNKD-MP-OKLAHOMA-AIRFLOW", "_microsoft_metrics_namespace": "airflow-${AIRFLOW_VAR_CLUSTER_NAME}", "Service.Name": "airflow", "Container.Name": "airflow-main-airflow-oklahoma:0.0.697", "Fabric": "${AIRFLOW_VAR_FABRIC}", "K8s.Pod.Name": "${POD_NAME}", "K8s.Namespace.Name": "${POD_NAMESPACE}", "Host.Name": "${HOSTNAME}" } AIRFLOW__METRICS__OTEL_EXPORTER_HEADERS: { "Content-Type": "application/json", "X-LI-Fluentbit-Tag":"serviceName:airflow,instanceId:i001,containerName:airflow-main-airflow-oklahoma:0.0.697,mdmAccount:LNKD-MP-OKLAHOMA-AIRFLOW,mdmNamespace:airflow-${AIRFLOW_VAR_CLUSTER_NAME}" } AIRFLOW__WEBSERVER__BASE_URL: <URL> AIRFLOW_VAR_FABRIC_GROUP: grid AIRFLOW_VAR_FABRIC: grid1 AIRFLOW_VAR_CLUSTER_NAME: load-test AIRFLOW__OKLAHOMA__GUNICORN_MAX_REQUESTS: 1150 AIRFLOW__OKLAHOMA__GUNICORN_MAX_REQUESTS_JITTER: 50 AIRFLOW__SCHEDULER__CATCHUP_BY_DEFAULT: False AIRFLOW__KUBERNETES_EXECUTOR__LABEL_SELECTORS: release=airflow-load-test AIRFLOW__CORE__PARALLELISM: 1024 AIRFLOW__SCHEDULER__SCHEDULER_HEALTH_CHECK_THRESHOLD: 180 AIRFLOW__CORE__MAX_ACTIVE_TASKS_PER_DAG: 2000 AIRFLOW__CORE__MAX_ACTIVE_RUNS_PER_DAG: 128 AIRFLOW__SCHEDULER__MAX_DAGRUNS_TO_CREATE_PER_LOOP: 15 AIRFLOW__SCHEDULER__MAX_DAGRUNS_PER_LOOP_TO_SCHEDULE: 20 AIRFLOW__SCHEDULER__MAX_TIS_PER_QUERY: 96 AIRFLOW__KUBERNETES_EXECUTOR__WORKER_PODS_CREATION_BATCH_SIZE: 50 AIRFLOW__KUBERNETES__DELETE_WORKER_PODS: True AIRFLOW__KUBERNETES__DELETE_WORKER_PODS_ON_FAILURE: True AIRFLOW__SCHEDULER__PARSING_PROCESSES: 8 AIRFLOW__SCHEDULER__MIN_FILE_PROCESS_INTERVAL: 900 AIRFLOW__SCHEDULER__DAG_DIR_LIST_INTERVAL: 420 AIRFLOW__CORE__DAG_FILE_PROCESSOR_TIMEOUT: 1200 AIRFLOW__SCHEDULER__STALE_DAG_THRESHOLD: 1200 AIRFLOW__CORE__DAGBAG_IMPORT_TIMEOUT: 120 AIRFLOW__SCHEDULER__MAX_ADOPTION_TIS_PER_QUERY: 50 AIRFLOW__KUBERNETES_EXECUTOR__MAX_ADOPTION_COMPLETED_PODS: 80 AIRFLOW__SCHEDULER__SCHEDULE_AFTER_TASK_EXECUTION: False DAG_ROOT_UPLOAD_ALLOW_LIST: service-multiproduct-post-commit,service-airflow-crt-action,service-darwin-platform-service,service-dsplatform-server-mt,service-airflow-main-airflow-oklahoma LOGGING_FALLBACK_TO_FILE_ENABLED: True AIRFLOW__LOGGING__LOGGING_CONFIG_CLASS: airflow.providers.lnkd.log.oklahoma_logging_config.OKLAHOMA_LOGGING_CONFIG AIRFLOW__LISTENER_PLUGIN__ENABLED: True AIRFLOW__KUBERNETES_ENVIRONMENT_VARIABLES__AIRFLOW__LISTENER_PLUGIN__ENABLED: True AIRFLOW_IS_K8S_EXECUTOR_POD: True Mounts: /etc/cm.json from cm-json (ro) /etc/lipki from lipki-certs (ro) /etc/riddler from riddler-certs (ro) /export/apps/oklahoma/azkaban from azkaban-account-key (ro) /export/apps/openssl/ssl/cert.pem from lnkd-cacert (ro) /opt/airflow/airflow.cfg from config (ro,path="airflow.cfg") /opt/airflow/config/airflow_local_settings.py from config (ro,path="airflow_local_settings.py") /opt/airflow/dags from dags (rw,path="load-test") /opt/airflow/logs from logs (rw) /var/cluster from dag-certs (rw) /var/cluster/mp from mp-certs (rw) /var/cluster/oklahoma from airflow-main-airflow-oklahoma-app-certs (rw) /var/run/secrets/kubernetes.io/serviceaccount from kube-api-access-jfss5 (ro) Conditions: Type Status Initialized True Ready False ContainersReady False PodScheduled True Volumes: dags: Type: PersistentVolumeClaim (a reference to a PersistentVolumeClaim in the same namespace) ClaimName: oklahoma-airflow-test-dag-pvc ReadOnly: false logs: Type: PersistentVolumeClaim (a reference to a PersistentVolumeClaim in the same namespace) ClaimName: oklahoma-airflow-test-log-pvc ReadOnly: false config: Type: ConfigMap (a volume populated by a ConfigMap) Name: airflow-load-test-config Optional: false riddler-certs: Type: HostPath (bare host directory volume) Path: /etc/riddler HostPathType: Directory airflow-main-airflow-oklahoma-app-certs: Type: EmptyDir (a temporary directory that shares a pod's lifetime) Medium: Memory SizeLimit: <unset> dag-certs: Type: EmptyDir (a temporary directory that shares a pod's lifetime) Medium: Memory SizeLimit: <unset> mp-certs: Type: EmptyDir (a temporary directory that shares a pod's lifetime) Medium: Memory SizeLimit: <unset> pki-agent-socket: Type: HostPath (bare host directory volume) Path: /var/run/lipki/pki-agent.sock HostPathType: Socket azkaban-account-key: Type: Secret (a volume populated by a Secret) SecretName: azkaban-account-key Optional: false cm-json: Type: HostPath (bare host directory volume) Path: /etc/cm.json HostPathType: File lnkd-cacert: Type: HostPath (bare host directory volume) Path: /etc/pki/ca-trust/extracted/pem/tls-ca-bundle.pem HostPathType: File lipki-certs: Type: HostPath (bare host directory volume) Path: /etc/lipki HostPathType: Directory kube-api-access-jfss5: Type: Projected (a volume that contains injected data from multiple sources) TokenExpirationSeconds: 3607 ConfigMapName: kube-root-ca.crt ConfigMapOptional: <nil> DownwardAPI: true QoS Class: Burstable Node-Selectors: <none> Tolerations: node.kubernetes.io/not-ready:NoExecute op=Exists for 300s node.kubernetes.io/unreachable:NoExecute op=Exists for 300s Events: <none> ``` ### 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]
