The Kubernetes executor has surprisingly low test coverage: 64% 

https://codecov.io/gh/apache/airflow/src/master/airflow/executors/kubernetes_executor.py - looks like some of the "flush/end" code is not tested.

We might want to improve it.
tests/executors/test_kubernetes_executor.py
mock_ti = mock.MagicMock(queued_by_job_id="1", external_executor_id="1", key=ti_key)
        pod = k8s.V1Pod(metadata=k8s.V1ObjectMeta(name="foo"))
        mock_kube_client = mock.MagicMock()
        mock_kube_client.list_namespaced_pod.return_value.items = [pod]
        executor.kube_client = mock_kube_client
        
        reset_tis = executor.try_adopt_task_instances([mock_ti])
        mock_kube_client.list_namespaced_pod.assert_called_once_with(
            namespace='default', label_selector='airflow-worker=1'
        )
        mock_adopt_launched_task.assert_called_once_with(mock_kube_client, pod, {ti_key: mock_ti})
        mock_adopt_completed_pods.assert_called_once()
        assert reset_tis == [mock_ti]  
        
        mock_kube_client.reset_mock()
        mock_adopt_launched_task.reset_mock()
        mock_adopt_completed_pods.reset_mock()
        mock_ti.queued_by_job_id = "10"  
        executor.scheduler_job_id = "20"
        
        mock_adopt_launched_task.side_effect = lambda client, pod, pod_ids: pod_ids.pop(ti_key)
        reset_tis = executor.try_adopt_task_instances([mock_ti])
        mock_kube_client.list_namespaced_pod.assert_called_once_with(
            namespace='default', label_selector='airflow-worker=10'
        )
        mock_adopt_launched_task.assert_called_once()  
        mock_adopt_completed_pods.assert_called_once()
        assert reset_tis == []  
    @mock.patch('airflow.executors.kubernetes_executor.KubernetesExecutor._adopt_completed_pods')
    def test_try_adopt_task_instances_multiple_scheduler_ids(self, mock_adopt_completed_pods):