Kubernetes api error in scheduler stops launching tasks

[iw-pavan]

Apache Airflow version

2.11.0

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

2.10.5

What happened?

There were connection issue on kubernetes api's where logs says terminating kubernetes executor. But watcher threads continue to run causing scheduler to not pick any taks

`Traceback (most recent call last):
File "/opt/app-root/lib64/python3.11/site-packages/urllib3/connectionpool.py", line 789, in urlopen
response = self._make_request(
^^^^^^^^^^^^^^^^^^^
File "/opt/app-root/lib64/python3.11/site-packages/urllib3/connectionpool.py", line 536, in _make_request
response = conn.getresponse()
^^^^^^^^^^^^^^^^^^
File "/opt/app-root/lib64/python3.11/site-packages/urllib3/connection.py", line 507, in getresponse
httplib_response = super().getresponse()
^^^^^^^^^^^^^^^^^^^^^
File "/usr/lib64/python3.11/http/client.py", line 1395, in getresponse
response.begin()
File "/usr/lib64/python3.11/http/client.py", line 325, in begin
version, status, reason = self._read_status()
^^^^^^^^^^^^^^^^^^^
File "/usr/lib64/python3.11/http/client.py", line 294, in _read_status
raise RemoteDisconnected("Remote end closed connection without"
http.client.RemoteDisconnected: Remote end closed connection without response

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
File "/opt/app-root/lib64/python3.11/site-packages/airflow/providers/cncf/kubernetes/executors/kubernetes_executor_utils.py", line 340, in run_pod_async
resp = self.kube_client.create_namespaced_pod(
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/opt/app-root/lib64/python3.11/site-packages/kubernetes/client/api/core_v1_api.py", line 7356, in create_namespaced_pod
return self.create_namespaced_pod_with_http_info(namespace, body, **kwargs) # noqa: E501
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/opt/app-root/lib64/python3.11/site-packages/kubernetes/client/api/core_v1_api.py", line 7455, in create_namespaced_pod_with_http_info
return self.api_client.call_api(
^^^^^^^^^^^^^^^^^^^^^^^^^
File "/opt/app-root/lib64/python3.11/site-packages/kubernetes/client/api_client.py", line 348, in call_api
return self.__call_api(resource_path, method,
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/opt/app-root/lib64/python3.11/site-packages/kubernetes/client/api_client.py", line 180, in __call_api
response_data = self.request(
^^^^^^^^^^^^^
File "/opt/app-root/lib64/python3.11/site-packages/kubernetes/client/api_client.py", line 391, in request
return self.rest_client.POST(url,
^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/opt/app-root/lib64/python3.11/site-packages/kubernetes/client/rest.py", line 279, in POST
return self.request("POST", url,
^^^^^^^^^^^^^^^^^^^^^^^^^
File "/opt/app-root/lib64/python3.11/site-packages/kubernetes/client/rest.py", line 172, in request
r = self.pool_manager.request(
^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/opt/app-root/lib64/python3.11/site-packages/urllib3/_request_methods.py", line 143, in request
return self.request_encode_body(
^^^^^^^^^^^^^^^^^^^^^^^^^
File "/opt/app-root/lib64/python3.11/site-packages/urllib3/_request_methods.py", line 278, in request_encode_body
return self.urlopen(method, url, **extra_kw)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/opt/app-root/lib64/python3.11/site-packages/urllib3/poolmanager.py", line 443, in urlopen
response = conn.urlopen(method, u.request_uri, **kw)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/opt/app-root/lib64/python3.11/site-packages/urllib3/connectionpool.py", line 843, in urlopen
retries = retries.increment(
^^^^^^^^^^^^^^^^^^
File "/opt/app-root/lib64/python3.11/site-packages/urllib3/util/retry.py", line 474, in increment
raise reraise(type(error), error, _stacktrace)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/opt/app-root/lib64/python3.11/site-packages/urllib3/util/util.py", line 38, in reraise
raise value.with_traceback(tb)
File "/opt/app-root/lib64/python3.11/site-packages/urllib3/connectionpool.py", line 789, in urlopen
response = self._make_request(
^^^^^^^^^^^^^^^^^^^
File "/opt/app-root/lib64/python3.11/site-packages/urllib3/connectionpool.py", line 536, in _make_request
response = conn.getresponse()
^^^^^^^^^^^^^^^^^^
File "/opt/app-root/lib64/python3.11/site-packages/urllib3/connection.py", line 507, in getresponse
httplib_response = super().getresponse()
^^^^^^^^^^^^^^^^^^^^^
File "/usr/lib64/python3.11/http/client.py", line 1395, in getresponse
response.begin()
File "/usr/lib64/python3.11/http/client.py", line 325, in begin
version, status, reason = self._read_status()
^^^^^^^^^^^^^^^^^^^
File "/usr/lib64/python3.11/http/client.py", line 294, in _read_status
raise RemoteDisconnected("Remote end closed connection without"
urllib3.exceptions.ProtocolError: ('Connection aborted.', RemoteDisconnected('Remote end closed connection without response'))
[2025-06-26T08:48:05.776+0000] {scheduler_job_runner.py:1016} ERROR - Exception when executing SchedulerJob._run_scheduler_loop
Traceback (most recent call last):
File "/opt/app-root/lib64/python3.11/site-packages/urllib3/connectionpool.py", line 789, in urlopen
response = self._make_request(
^^^^^^^^^^^^^^^^^^^
File "/opt/app-root/lib64/python3.11/site-packages/urllib3/connectionpool.py", line 536, in _make_request
response = conn.getresponse()
^^^^^^^^^^^^^^^^^^
File "/opt/app-root/lib64/python3.11/site-packages/urllib3/connection.py", line 507, in getresponse
httplib_response = super().getresponse()
^^^^^^^^^^^^^^^^^^^^^
File "/usr/lib64/python3.11/http/client.py", line 1395, in getresponse
response.begin()
File "/usr/lib64/python3.11/http/client.py", line 325, in begin
version, status, reason = self._read_status()
^^^^^^^^^^^^^^^^^^^
File "/usr/lib64/python3.11/http/client.py", line 294, in _read_status
raise RemoteDisconnected("Remote end closed connection without"
http.client.RemoteDisconnected: Remote end closed connection without response

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
File "/opt/app-root/lib64/python3.11/site-packages/airflow/jobs/scheduler_job_runner.py", line 999, in _execute
self._run_scheduler_loop()
File "/opt/app-root/lib64/python3.11/site-packages/airflow/jobs/scheduler_job_runner.py", line 1144, in _run_scheduler_loop
executor.heartbeat()
File "/opt/app-root/lib64/python3.11/site-packages/airflow/traces/tracer.py", line 58, in wrapper
return func(*args, **kwargs)
^^^^^^^^^^^^^^^^^^^^^
File "/opt/app-root/lib64/python3.11/site-packages/airflow/executors/base_executor.py", line 248, in heartbeat
self.sync()
File "/opt/app-root/lib64/python3.11/site-packages/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py", line 422, in sync
self.kube_scheduler.run_next(task)
File "/opt/app-root/lib64/python3.11/site-packages/airflow/providers/cncf/kubernetes/executors/kubernetes_executor_utils.py", line 435, in run_next
self.run_pod_async(pod, **self.kube_config.kube_client_request_args)
File "/opt/app-root/lib64/python3.11/site-packages/airflow/providers/cncf/kubernetes/executors/kubernetes_executor_utils.py", line 346, in run_pod_async
raise e
File "/opt/app-root/lib64/python3.11/site-packages/airflow/providers/cncf/kubernetes/executors/kubernetes_executor_utils.py", line 340, in run_pod_async
resp = self.kube_client.create_namespaced_pod(
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/opt/app-root/lib64/python3.11/site-packages/kubernetes/client/api/core_v1_api.py", line 7356, in create_namespaced_pod
return self.create_namespaced_pod_with_http_info(namespace, body, **kwargs) # noqa: E501
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/opt/app-root/lib64/python3.11/site-packages/kubernetes/client/api/core_v1_api.py", line 7455, in create_namespaced_pod_with_http_info
return self.api_client.call_api(
^^^^^^^^^^^^^^^^^^^^^^^^^
File "/opt/app-root/lib64/python3.11/site-packages/kubernetes/client/api_client.py", line 348, in call_api
return self.__call_api(resource_path, method,
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/opt/app-root/lib64/python3.11/site-packages/kubernetes/client/api_client.py", line 180, in __call_api
response_data = self.request(
^^^^^^^^^^^^^
File "/opt/app-root/lib64/python3.11/site-packages/kubernetes/client/api_client.py", line 391, in request
return self.rest_client.POST(url,
^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/opt/app-root/lib64/python3.11/site-packages/kubernetes/client/rest.py", line 279, in POST
return self.request("POST", url,
^^^^^^^^^^^^^^^^^^^^^^^^^
File "/opt/app-root/lib64/python3.11/site-packages/kubernetes/client/rest.py", line 172, in request
r = self.pool_manager.request(
^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/opt/app-root/lib64/python3.11/site-packages/urllib3/_request_methods.py", line 143, in request
return self.request_encode_body(
^^^^^^^^^^^^^^^^^^^^^^^^^
File "/opt/app-root/lib64/python3.11/site-packages/urllib3/_request_methods.py", line 278, in request_encode_body
return self.urlopen(method, url, **extra_kw)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/opt/app-root/lib64/python3.11/site-packages/urllib3/poolmanager.py", line 443, in urlopen
response = conn.urlopen(method, u.request_uri, **kw)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/opt/app-root/lib64/python3.11/site-packages/urllib3/connectionpool.py", line 843, in urlopen
retries = retries.increment(
^^^^^^^^^^^^^^^^^^
File "/opt/app-root/lib64/python3.11/site-packages/urllib3/util/retry.py", line 474, in increment
raise reraise(type(error), error, _stacktrace)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/opt/app-root/lib64/python3.11/site-packages/urllib3/util/util.py", line 38, in reraise
raise value.with_traceback(tb)
File "/opt/app-root/lib64/python3.11/site-packages/urllib3/connectionpool.py", line 789, in urlopen
response = self._make_request(
^^^^^^^^^^^^^^^^^^^
File "/opt/app-root/lib64/python3.11/site-packages/urllib3/connectionpool.py", line 536, in _make_request
response = conn.getresponse()
^^^^^^^^^^^^^^^^^^
File "/opt/app-root/lib64/python3.11/site-packages/urllib3/connection.py", line 507, in getresponse
httplib_response = super().getresponse()
^^^^^^^^^^^^^^^^^^^^^
File "/usr/lib64/python3.11/http/client.py", line 1395, in getresponse
response.begin()
File "/usr/lib64/python3.11/http/client.py", line 325, in begin
version, status, reason = self._read_status()
^^^^^^^^^^^^^^^^^^^
File "/usr/lib64/python3.11/http/client.py", line 294, in _read_status
raise RemoteDisconnected("Remote end closed connection without"
urllib3.exceptions.ProtocolError: ('Connection aborted.', RemoteDisconnected('Remote end closed connection without response'))
[2025-06-26T08:48:05.835+0000] {kubernetes_executor.py:783} INFO - Shutting down Kubernetes executor
[2025-06-26T08:48:31.808+0000] {kubernetes_executor_utils.py:95} INFO - Kubernetes watch timed out waiting for events. Restarting watch.
[2025-06-26T08:48:32.814+0000] {kubernetes_executor_utils.py:134} INFO - Event: and now my watch begins starting at resource_version: 0
[2025-06-26T08:48:32.814+0000] {kubernetes_executor_utils.py:134} INFO - Event: and now my watch begins starting at resource_version: 0
[2025-06-26T08:48:32.864+0000] {kubernetes_executor_utils.py:278} ERROR - Event: 685aa942b741b073e17a652b-bs-8401-jcj0m36g Failed, annotations:
[2025-06-26T08:49:02.893+0000] {kubernetes_executor_utils.py:95} INFO - Kubernetes watch timed out waiting for events. Restarting watch.
[2025-06-26T08:49:03.900+0000] {kubernetes_executor_utils.py:134} INFO - Event: and now my watch begins starting at resource_version: 0
[2025-06-26T08:49:03.964+0000] {kubernetes_executor_utils.py:278} ERROR - Event: 685aa942b741b073e17a652b-bs-8401-jcj0m36g Failed, annotations:
[2025-06-26T08:49:33.986+0000] {kubernetes_executor_utils.py:95} INFO - Kubernetes watch timed out waiting for events. Restarting watch.
[2025-06-26T08:49:34.993+0000] {kubernetes_executor_utils.py:134} INFO - Event: and now my watch begins starting at resource_version: 0
[2025-06-26T08:49:35.059+0000] {kubernetes_executor_utils.py:278} ERROR - Event: 685aa942b741b073e17a652b-bs-8401-jcj0m36g Failed, annotations:
[2025-06-26T08:50:05.080+0000] {kubernetes_executor_utils.py:95} INFO - Kubernetes watch timed out waiting for events. Restarting watch.
[2025-06-26T08:50:06.087+0000] {kubernetes_executor_utils.py:134} INFO - Event: and now my watch begins starting at resource_version: 0
[2025-06-26T08:50:06.145+0000] {kubernetes_executor_utils.py:278} ERROR - Event: 685aa942b741b073e17a652b-bs-8401-jcj0m36g Failed, annotations:
[2025-06-26T08:50:36.164+0000] {kubernetes_executor_utils.py:95} INFO - Kubernetes watch timed out waiting for events. Restarting watch.
[2025-06-26T08:50:37.173+0000] {kubernetes_executor_utils.py:134} INFO - Event: and now my watch begins starting at resource_version: 0
[2025-06-26T08:50:37.218+0000] {kubernetes_executor_utils.py:278} ERROR - Event: 685aa942b741b073e17a652b-bs-8401-jcj0m36g Failed, annotations:
[2025-06-26T08:51:07.244+0000] {kubernetes_executor_utils.py:95} INFO - Kubernetes watch timed out waiting for events. Restarting watch.
[2025-06-26T08:51:08.251+0000] {kubernetes_executor_utils.py:134} INFO - Event: and now my watch begins starting at resource_version: 0
[2025-06-26T08:51:08.305+0000] {kubernetes_executor_utils.py:278} ERROR - Event: 685aa942b741b073e17a652b-bs-8401-jcj0m36g Failed, annotations:
[2025-06-26T08:51:38.328+0000] {kubernetes_executor_utils.py:95} INFO - Kubernetes watch timed out waiting for events. Restarting watch.
[2025-06-26T08:51:39.334+0000] {kubernetes_executor_utils.py:134} INFO - Event: and now my watch begins starting at resource_version: 0
[2025-06-26T08:51:39.408+0000] {kubernetes_executor_utils.py:278} ERROR - Event: 685aa942b741b073e17a652b-bs-8401-jcj0m36g Failed, annotations:
[2025-06-26T08:52:09.414+0000] {kubernetes_executor_utils.py:95} INFO - Kubernetes watch timed out waiting for events. Restarting watch.
[2025-06-26T08:52:10.421+0000] {kubernetes_executor_utils.py:134} INFO - Event: and now my watch begins starting at resource_version: 0
[2025-06-26T08:52:10.479+0000] {kubernetes_executor_utils.py:278} ERROR - Event: 685aa942b741b073e17a652b-bs-8401-jcj0m36g Failed, annotations:
[2025-06-26T08:52:40.508+0000] {kubernetes_executor_utils.py:95} INFO - Kubernetes watch timed out waiting for events. Restarting watch.
[2025-06-26T08:52:41.515+0000] {kubernetes_executor_utils.py:134} INFO - Event: and now my watch begins starting at resource_version: 0
[2025-06-26T08:52:41.570+0000] {kubernetes_executor_utils.py:278} ERROR - Event: 685aa942b741b073e17a652b-bs-8401-jcj0m36g Failed, annotations:
[2025-06-26T08:53:11.593+0000] {kubernetes_executor_utils.py:95} INFO - Kubernetes watch timed out waiting for events. Restarting watch.
[2025-06-26T08:53:12.599+0000] {kubernetes_executor_utils.py:134} INFO - Event: and now my watch begins starting at resource_version: 0
[2025-06-26T08:53:12.639+0000] {kubernetes_executor_utils.py:278} ERROR - Event: 685aa942b741b073e17a652b-bs-8401-jcj0m36g Failed, annotations:
[2025-06-26T08:53:42.664+0000] {kubernetes_executor_utils.py:95} INFO - Kubernetes watch timed out waiting for events. Restarting watch.
[2025-06-26T08:53:43.670+0000] {kubernetes_executor_utils.py:134} INFO - Event: and now my watch begins starting at resource_version: 0
[2025-06-26T08:53:43.724+0000] {kubernetes_executor_utils.py:278} ERROR - Event: 685aa942b741b073e17a652b-bs-8401-jcj0m36g Failed, annotations:
[2025-06-26T08:54:13.745+0000] {kubernetes_executor_utils.py:95} INFO - Kubernetes watch timed out waiting for events. Restarting watch.
[2025-06-26T08:54:14.751+0000] {kubernetes_executor_utils.py:134} INFO - Event: and now my watch begins starting at resource_version: 0
[2025-06-26T08:54:14.800+0000] {kubernetes_executor_utils.py:278} ERROR - Event: 685aa942b741b073e17a652b-bs-8401-jcj0m36g Failed, annotations:
[2025-06-26T08:54:44.830+0000] {kubernetes_executor_utils.py:95} INFO - Kubernetes watch timed out waiting for events. Restarting watch.
[2025-06-26T08:54:45.837+0000] {kubernetes_executor_utils.py:134} INFO - Event: and now my watch begins starting at resource_version: 0
[2025-06-26T08:54:45.890+0000] {kubernetes_executor_utils.py:278} ERROR - Event: 685aa942b741b073e17a652b-bs-8401-jcj0m36g Failed, annotations:
[2025-06-26T08:55:15.916+0000] {kubernetes_executor_utils.py:95} INFO - Kubernetes watch timed out waiting for events. Restarting watch.
[2025-06-26T08:55:16.924+0000] {kubernetes_executor_utils.py:134} INFO - Event: and now my watch begins starting at resource_version: 0
[2025-06-26T08:55:16.979+0000] {kubernetes_executor_utils.py:278} ERROR - Event: 685aa942b741b073e17a652b-bs-8401-jcj0m36g Failed, annotations:
[2025-06-26T08:55:47.000+0000] {kubernetes_executor_utils.py:95} INFO - Kubernetes watch timed out waiting for events. Restarting watch.
[2025-06-26T08:55:48.006+0000] {kubernetes_executor_utils.py:134} INFO - Event: and now my watch begins starting at resource_version: 0
[2025-06-26T08:55:48.060+0000] {kubernetes_executor_utils.py:278} ERROR - Event: 685aa942b741b073e17a652b-bs-8401-jcj0m36g Failed, annotations:
[2025-06-26T08:56:18.080+0000] {kubernetes_executor_utils.py:95} INFO - Kubernetes watch timed out waiting for events. Restarting watch.
[2025-06-26T08:56:19.089+0000] {kubernetes_executor_utils.py:134} INFO - Event: and now my watch begins starting at resource_version: 0
[2025-06-26T08:56:19.150+0000] {kubernetes_executor_utils.py:278} ERROR - Event: 685aa942b741b073e17a652b-bs-8401-jcj0m36g Failed, annotations:
[2025-06-26T08:56:49.166+0000] {kubernetes_executor_utils.py:95} INFO - Kubernetes watch timed out waiting for events. Restarting watch.`

What you think should happen instead?

No response

How to reproduce

Create db errors or api error on kubernetes api instance

Operating System

Linux

Versions of Apache Airflow Providers

No response

Deployment

Docker-Compose

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

[iw-pavan]

Fix for #43440 seems have some edge case

[iw-pavan]

Since kubewatcher is while True loop, it might not terminate
should I add force kill if it doesnt terminate?

ie

for kube_watcher in self.kube_watchers.values():
    kube_watcher.terminate()
    kube_watcher.join(timeout=60)
    if kube_watcher.is_alive():
        self.log.warning("KubeJobWatcher did not terminate in time. killing it.")
        kube_watcher.kill()
        kube_watcher.join()
    self.log.debug("kube_watcher=%s", kube_watcher)

[wherrmann]

@iw-pavan this isn't a long-term fix but as a heads up — we experienced this yesterday and it cleared up after restarting the scheduler pod(s).

[iw-pavan]

@wherrmann
we are noticing this once a month or so, restarting scheduler fix it, but it adds manual action.

[kosteev]

I can confirm that I can see this issue with Airflow 3.0.0 (CeleryExecutor+KubernetesExecutor) - scheduler is not exiting after reaching limit number of loops, continues to watch k8s events.