CeleryKubernetesExecutor does not shutdown gracefully in 2.2.0

[dstandish]

Apache Airflow version

2.2.0 (latest released)

Operating System

debian

Deployment

Official Apache Airflow Helm Chart

Running CeleryKubernetesExecutor

What happened

I noticed that scheduler was restarting very frequently after upgrading to 2.2.0. It turns out this was due to the slow liveness probe (see #19001).

And when I looked at the logs I found that when kubernetes executor tried to shut down, it errored out.

[2021-10-14 16:00:45,192] {scheduler_job.py:161} INFO - Exiting gracefully upon receiving signal 15
[2021-10-14 16:00:46,194] {process_utils.py:100} INFO - Sending Signals.SIGTERM to GPID 33
[2021-10-14 16:00:46,673] {process_utils.py:66} INFO - Process psutil.Process(pid=33, status='terminated', exitcode=0, started='15:56:00') (33) terminated with exit code 0
[2021-10-14 16:00:46,741] {kubernetes_executor.py:787} INFO - Shutting down Kubernetes executor
[2021-10-14 16:00:46,741] {scheduler_job.py:609} ERROR - Exception when executing Executor.end
Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 587, in _execute
    self._run_scheduler_loop()
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 689, in _run_scheduler_loop
    time.sleep(min(self._scheduler_idle_sleep_time, next_event))
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 164, in _exit_gracefully
    sys.exit(os.EX_OK)
SystemExit: 0

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 607, in _execute
    self.executor.end()
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/executors/celery_kubernetes_executor.py", line 177, in end
    self.kubernetes_executor.end()
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/executors/kubernetes_executor.py", line 789, in end
    self._flush_task_queue()
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/executors/kubernetes_executor.py", line 742, in _flush_task_queue
    self.log.debug('Executor shutting down, task_queue approximate size=%d', self.task_queue.qsize())
  File "<string>", line 2, in qsize
  File "/usr/local/lib/python3.8/multiprocessing/managers.py", line 834, in _callmethod
    conn.send((self._id, methodname, args, kwds))
  File "/usr/local/lib/python3.8/multiprocessing/connection.py", line 206, in send
    self._send_bytes(_ForkingPickler.dumps(obj))
  File "/usr/local/lib/python3.8/multiprocessing/connection.py", line 411, in _send_bytes
    self._send(header + buf)
  File "/usr/local/lib/python3.8/multiprocessing/connection.py", line 368, in _send
    n = write(self._handle, buf)
BrokenPipeError: [Errno 32] Broken pipe
[2021-10-14 16:00:46,744] {process_utils.py:100} INFO - Sending Signals.SIGTERM to GPID 33
[2021-10-14 16:00:46,744] {scheduler_job.py:614} INFO - Exited execute loop

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

[seancaulfield]

We also have this issue, but with the KubernetesExecutor. Airflow version is 2.1.4. Seems to happen whenever the scheduler is being restarted by the deployment.

Here is the stack trace, but it's likely a little jumbled as I had to pull it from DataDog and the lines don't always arrive in order there:

[2021-11-18 17:24:30,593] {process_utils.py:100} INFO - Sending Signals.SIGTERM to GPID 60
[2021-11-18 17:24:30,593] {scheduler_job.py:722} INFO - Exited execute loop
Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/executors/kubernetes_executor.py", line 552, in sync
    self._change_state(key, state, pod_id, namespace)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/executors/kubernetes_executor.py", line 634, in _change_state
    self.kube_scheduler.delete_pod(pod_id, namespace)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/executors/kubernetes_executor.py", line 340, in delete_pod
    **self.kube_config.kube_client_request_args,
  File "/home/airflow/.local/lib/python3.7/site-packages/kubernetes/client/api/core_v1_api.py", line 10173, in delete_namespaced_pod
    (data) = self.delete_namespaced_pod_with_http_info(name, namespace, **kwargs)  # noqa: E501
  File "/home/airflow/.local/lib/python3.7/site-packages/kubernetes/client/api/core_v1_api.py", line 10273, in delete_namespaced_pod_with_http_info
    collection_formats=collection_formats)
  File "/home/airflow/.local/lib/python3.7/site-packages/kubernetes/client/api_client.py", line 345, in call_api
    _preload_content, _request_timeout)
  File "/home/airflow/.local/lib/python3.7/site-packages/kubernetes/client/api_client.py", line 176, in __call_api
    _request_timeout=_request_timeout)
  File "/home/airflow/.local/lib/python3.7/site-packages/kubernetes/client/api_client.py", line 411, in request
    body=body)
  File "/home/airflow/.local/lib/python3.7/site-packages/kubernetes/client/rest.py", line 268, in DELETE
    body=body)
  File "/home/airflow/.local/lib/python3.7/site-packages/kubernetes/client/rest.py", line 170, in request
    headers=headers)
  File "/home/airflow/.local/lib/python3.7/site-packages/urllib3/request.py", line 75, in request
    method, url, fields=fields, headers=headers, **urlopen_kw
  File "/home/airflow/.local/lib/python3.7/site-packages/urllib3/request.py", line 96, in request_encode_url
    return self.urlopen(method, url, **extra_kw)
  File "/home/airflow/.local/lib/python3.7/site-packages/urllib3/poolmanager.py", line 375, in urlopen
    response = conn.urlopen(method, u.request_uri, **kw)
  File "/home/airflow/.local/lib/python3.7/site-packages/urllib3/connectionpool.py", line 706, in urlopen
    chunked=chunked,
  File "/home/airflow/.local/lib/python3.7/site-packages/urllib3/connectionpool.py", line 445, in _make_request
    six.raise_from(e, None)
  File "<string>", line 3, in raise_from
  File "/home/airflow/.local/lib/python3.7/site-packages/urllib3/connectionpool.py", line 440, in _make_request
    httplib_response = conn.getresponse()
  File "/usr/local/lib/python3.7/http/client.py", line 1373, in getresponse
    response.begin()
  File "/usr/local/lib/python3.7/http/client.py", line 319, in begin
    version, status, reason = self._read_status()
  File "/usr/local/lib/python3.7/http/client.py", line 280, in _read_status
    line = str(self.fp.readline(_MAXLINE + 1), "iso-8859-1")
  File "/usr/local/lib/python3.7/socket.py", line 589, in readinto
    return self._sock.recv_into(b)
  File "/usr/local/lib/python3.7/ssl.py", line 1071, in recv_into
    return self.read(nbytes, buffer)
  File "/usr/local/lib/python3.7/ssl.py", line 929, in read
    return self._sslobj.read(len, buffer)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/jobs/scheduler_job.py", line 155, in _exit_gracefully
    sys.exit(os.EX_OK)
SystemExit: 0

During handling of the above exception, another exception occurred:
Traceback (most recent call last):
  File "/home/airflow/.local/bin/airflow", line 8, in <module>
    sys.exit(main())
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/__main__.py", line 40, in main
    args.func(args)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/cli/cli_parser.py", line 48, in command
    return func(*args, **kwargs)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/utils/cli.py", line 92, in wrapper
    return f(*args, **kwargs)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/cli/commands/scheduler_command.py", line 70, in scheduler
    job.run()
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/jobs/base_job.py", line 245, in run
    self._execute()
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/jobs/scheduler_job.py", line 695, in _execute
    self._run_scheduler_loop()
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/jobs/scheduler_job.py", line 790, in _run_scheduler_loop
    self.executor.heartbeat()
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/executors/base_executor.py", line 162, in heartbeat
    self.sync()
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/executors/kubernetes_executor.py", line 562, in sync
    self.result_queue.task_done()
  File "<string>", line 2, in task_done
  File "/usr/local/lib/python3.7/multiprocessing/managers.py", line 818, in _callmethod
    conn.send((self._id, methodname, args, kwds))
  File "/usr/local/lib/python3.7/multiprocessing/connection.py", line 206, in send
    self._send_bytes(_ForkingPickler.dumps(obj))
  File "/usr/local/lib/python3.7/multiprocessing/connection.py", line 404, in _send_bytes
    self._send(header + buf)
  File "/usr/local/lib/python3.7/multiprocessing/connection.py", line 368, in _send
    n = write(self._handle, buf)
BrokenPipeError: [Errno 32] Broken pipe

[NikolaySokolov]

We are seeing it on apache/airflow:2.2.3-python3.8. Any advice on how to troubleshoot it?

[2022-01-06 15:08:44,900] {kubernetes_executor.py:147} INFO - Event: airflowdbcleanupprintconfiguration.a31e0e08678046d1878d5419d3eb70eb had an event of type MODIFIED
[2022-01-06 15:08:44,901] {kubernetes_executor.py:201} INFO - Event: airflowdbcleanupprintconfiguration.a31e0e08678046d1878d5419d3eb70eb Pending
[2022-01-06 15:08:45,898] {kubernetes_executor.py:147} INFO - Event: airflowdbcleanupprintconfiguration.a31e0e08678046d1878d5419d3eb70eb had an event of type MODIFIED
[2022-01-06 15:08:45,899] {kubernetes_executor.py:213} INFO - Event: airflowdbcleanupprintconfiguration.a31e0e08678046d1878d5419d3eb70eb is Running
[2022-01-06 15:08:56,092] {scheduler_job.py:161} INFO - Exiting gracefully upon receiving signal 15
[2022-01-06 15:08:57,097] {process_utils.py:120} INFO - Sending Signals.SIGTERM to group 55. PIDs of all processes in the group: [4659, 4664, 4670, 4677, 4680, 4681, 55]
[2022-01-06 15:08:57,098] {process_utils.py:75} INFO - Sending the signal Signals.SIGTERM to group 55
[2022-01-06 15:09:02,433] {process_utils.py:70} INFO - Process psutil.Process(pid=4659, status='terminated', started='15:08:54') (4659) terminated with exit code None
[2022-01-06 15:09:02,434] {process_utils.py:70} INFO - Process psutil.Process(pid=4670, status='terminated', started='15:08:54') (4670) terminated with exit code None
[2022-01-06 15:09:02,519] {process_utils.py:70} INFO - Process psutil.Process(pid=4680, status='terminated', started='15:08:55') (4680) terminated with exit code None
[2022-01-06 15:09:02,607] {process_utils.py:237} INFO - Waiting up to 5 seconds for processes to exit...
Exception ignored in: <function _releaseLock at 0x7f3d1b320940>
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/logging/__init__.py", line 227, in _releaseLock
    def _releaseLock():
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/dag_processing/manager.py", line 500, in _exit_gracefully
    sys.exit(os.EX_OK)
SystemExit: 0
Process ForkProcess-4:
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/multiprocessing/process.py", line 315, in _bootstrap
    self.run()
  File "/usr/local/lib/python3.8/multiprocessing/process.py", line 108, in run
    self._target(*self._args, **self._kwargs)
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/dag_processing/manager.py", line 287, in _run_processor_manager
    processor_manager.start()
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/dag_processing/manager.py", line 520, in start
    return self._run_parsing_loop()
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/dag_processing/manager.py", line 609, in _run_parsing_loop
    self.collect_results()
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/dag_processing/manager.py", line 935, in collect_results
    self._collect_results_from_processor(processor)
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/dag_processing/manager.py", line 904, in _collect_results_from_processor
    if processor.result is not None:
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/dag_processing/processor.py", line 322, in result
    if not self.done:
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/dag_processing/processor.py", line 287, in done
    if self._parent_channel.poll():
  File "/usr/local/lib/python3.8/multiprocessing/connection.py", line 255, in poll
    self._check_closed()
  File "/usr/local/lib/python3.8/multiprocessing/connection.py", line 136, in _check_closed
    raise OSError("handle is closed")
OSError: handle is closed
[2022-01-06 15:09:02,652] {process_utils.py:70} INFO - Process psutil.Process(pid=55, status='terminated', exitcode=1, started='15:06:11') (55) terminated with exit code 1
[2022-01-06 15:09:02,653] {process_utils.py:70} INFO - Process psutil.Process(pid=4664, status='terminated', started='15:08:54') (4664) terminated with exit code None
[2022-01-06 15:09:02,653] {process_utils.py:70} INFO - Process psutil.Process(pid=4681, status='terminated', started='15:08:56') (4681) terminated with exit code None
[2022-01-06 15:09:02,653] {process_utils.py:70} INFO - Process psutil.Process(pid=4677, status='terminated', started='15:08:55') (4677) terminated with exit code None
[2022-01-06 15:09:02,654] {kubernetes_executor.py:790} INFO - Shutting down Kubernetes executor
[2022-01-06 15:09:02,654] {scheduler_job.py:650} ERROR - Exception when executing Executor.end
Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 628, in _execute
    self._run_scheduler_loop()
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 730, in _run_scheduler_loop
    time.sleep(min(self._scheduler_idle_sleep_time, next_event))
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 164, in _exit_gracefully
    sys.exit(os.EX_OK)
SystemExit: 0

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 648, in _execute
    self.executor.end()
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/executors/kubernetes_executor.py", line 792, in end
    self._flush_task_queue()
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/executors/kubernetes_executor.py", line 745, in _flush_task_queue
    self.log.debug('Executor shutting down, task_queue approximate size=%d', self.task_queue.qsize())
  File "<string>", line 2, in qsize
  File "/usr/local/lib/python3.8/multiprocessing/managers.py", line 834, in _callmethod
    conn.send((self._id, methodname, args, kwds))
  File "/usr/local/lib/python3.8/multiprocessing/connection.py", line 206, in send
    self._send_bytes(_ForkingPickler.dumps(obj))
  File "/usr/local/lib/python3.8/multiprocessing/connection.py", line 411, in _send_bytes
    self._send(header + buf)
  File "/usr/local/lib/python3.8/multiprocessing/connection.py", line 368, in _send
    n = write(self._handle, buf)
BrokenPipeError: [Errno 32] Broken pipe
[2022-01-06 15:09:02,656] {process_utils.py:120} INFO - Sending Signals.SIGTERM to group 55. PIDs of all processes in the group: []
[2022-01-06 15:09:02,657] {process_utils.py:75} INFO - Sending the signal Signals.SIGTERM to group 55
[2022-01-06 15:09:02,657] {process_utils.py:89} INFO - Sending the signal Signals.SIGTERM to process 55 as process group is missing.
[2022-01-06 15:09:02,657] {scheduler_job.py:655} INFO - Exited execute loop

[rtudo]

Facing the similar issue.

[hamedhsn]

Having the same issue. Seeing the problem after we pulled the latest version of the image(apache/airflow:2.2.2-python3.9) which was pushed a month ago.
previous to that we were using the image with the same tag and it was fine. something has been introduced in the new image I think. we are using kubernetes executor btw.

[potiuk]

We are going to switch to latest Celery 5.2.3 in 2.2.4 - can you please take a look (it solves a number of stabilty problems in Celery). RC should be out some time next week most likely.

[rmakoto-ze]

Same issue here after upgrading to apache/airflow:2.2.3. We are also using Kubernetes Executor.
Any idea of what is causing the problem?

[nanaones]

Facing same issue
helm chart 1.3.0
image version apache/airflow:2.2.2