AIRFLOW-5581: Join KubernetesJobWatcher in terminate call and unblock queues from blocking forever

[kpathak13]

Make sure you have checked all steps below.

Jira

• My PR addresses the following Airflow Jira issues and references them in the PR:
  • https://issues.apache.org/jira/browse/AIRFLOW-5581

Description

• It fixes a bug in kubernetes_executor module

If run_duration or num_runs is enabled, KubernetesJobWatcher dies prematurely causing below error:

[2019-10-01 19:40:35,323] {kubernetes_executor.py:327} ERROR - Unknown error in KubernetesJobWatcher. Failing
Traceback (most recent call last):
File "/home/vccorp/.local/lib/python3.6/site-packages/airflow/contrib/executors/kubernetes_executor.py", line 325, in run
self.worker_uuid, self.kube_config)
File "/home/vccorp/.local/lib/python3.6/site-packages/airflow/contrib/executors/kubernetes_executor.py", line 359, in _run
task.metadata.resource_version
File "/home/vccorp/.local/lib/python3.6/site-packages/airflow/contrib/executors/kubernetes_executor.py", line 391, in process_status
self.watcher_queue.put((pod_id, None, labels, resource_version))
File "", line 2, in put
File "/usr/lib64/python3.6/multiprocessing/managers.py", line 756, in _callmethod
conn.send((self._id, methodname, args, kwds))
File "/usr/lib64/python3.6/multiprocessing/connection.py", line 206, in send
self._send_bytes(_ForkingPickler.dumps(obj))
File "/usr/lib64/python3.6/multiprocessing/connection.py", line 404, in _send_bytes
self._send(header + buf)
File "/usr/lib64/python3.6/multiprocessing/connection.py", line 368, in _send
n = write(self._handle, buf)
BrokenPipeError: [Errno 32] Broken pipe
Process KubernetesJobWatcher-3:
Traceback (most recent call last):
File "/usr/lib64/python3.6/multiprocessing/process.py", line 258, in _bootstrap
self.run()
File "/home/vccorp/.local/lib/python3.6/site-packages/airflow/contrib/executors/kubernetes_executor.py", line 325, in run
self.worker_uuid, self.kube_config)
File "/home/vccorp/.local/lib/python3.6/site-packages/airflow/contrib/executors/kubernetes_executor.py", line 359, in _run
task.metadata.resource_version
File "/home/vccorp/.local/lib/python3.6/site-packages/airflow/contrib/executors/kubernetes_executor.py", line 391, in process_status
self.watcher_queue.put((pod_id, None, labels, resource_version))
File "", line 2, in put
File "/usr/lib64/python3.6/multiprocessing/managers.py", line 756, in _callmethod
conn.send((self._id, methodname, args, kwds))
File "/usr/lib64/python3.6/multiprocessing/connection.py", line 206, in send
self._send_bytes(_ForkingPickler.dumps(obj))
File "/usr/lib64/python3.6/multiprocessing/connection.py", line 404, in _send_bytes
self._send(header + buf)
File "/usr/lib64/python3.6/multiprocessing/connection.py", line 368, in _send
n = write(self._handle, buf)
BrokenPipeError: [Errno 32] Broken pipe

Tests

• This does not have test case since it's a single line update.

[codecov-io]

Codecov Report

Merging #6237 into master will decrease coverage by 0.15%.
The diff coverage is 6.66%.

@@            Coverage Diff             @@
##           master    #6237      +/-   ##
==========================================
- Coverage   80.04%   79.89%   -0.16%     
==========================================
  Files         610      610              
  Lines       35264    35306      +42     
==========================================
- Hits        28227    28207      -20     
- Misses       7037     7099      +62

Impacted Files	Coverage Δ
airflow/executors/kubernetes_executor.py	58.89% <6.66%> (-6.34%)	⬇️
airflow/operators/postgres_operator.py	0% <0%> (-100%)	⬇️
airflow/hooks/postgres_hook.py	94.73% <0%> (-1.76%)	⬇️
airflow/utils/sqlalchemy.py	91.52% <0%> (-1.7%)	⬇️
airflow/hooks/dbapi_hook.py	86.44% <0%> (-1.7%)	⬇️
airflow/models/baseoperator.py	95.27% <0%> (ø)	⬆️
airflow/bin/cli.py	66.38% <0%> (+0.2%)	⬆️
airflow/models/taskinstance.py	93.75% <0%> (+0.5%)	⬆️

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 18c62da...55ba865. Read the comment docs.

[dimberman]

@kpathak13 LGTM. Can you rebase? I'll merge when ready :)

[ashb]

This isn't useful - after a call to .join() the kube_watcher should never be alive, right?

[kpathak13]

Removed is alive? and rebased.

[ashb]

The code looks okay, but the important question: how have you tested this?

[kpathak13]

@ashb , I tested this on my k8 cluster:

1. Scheduled multiple high frequency jobs 
2. Set Scheduler run duration to 2 minutes
3. Killed several pods in flight at different stages (multiple times)
4. Deleted Scheduler pod

Never observed race condition.