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
executor = self.kubernetes_executor
        executor.heartbeat()
        calls = [
            mock.call('executor.open_slots', mock.ANY),
            mock.call('executor.queued_tasks', mock.ANY),
            mock.call('executor.running_tasks', mock.ANY),
        ]
        mock_stats_gauge.assert_has_calls(calls)
    @mock.patch('airflow.executors.kubernetes_executor.KubernetesJobWatcher')
    @mock.patch('airflow.executors.kubernetes_executor.get_kube_client')
    def test_invalid_executor_config(self, mock_get_kube_client, mock_kubernetes_job_watcher):
        executor = self.kubernetes_executor
        executor.start()
        assert executor.event_buffer == {}
        executor.execute_async(
            key=('dag', 'task', datetime.utcnow(), 1),
            queue=None,
            command=['airflow', 'tasks', 'run', 'true', 'some_parameter'],
            executor_config=k8s.V1Pod(
                spec=k8s.V1PodSpec(
                    containers=[k8s.V1Container(name="base", image="myimage", image_pull_policy="Always")]
                )
            ),
        )
        assert list(executor.event_buffer.values())[0][1] == "Invalid executor_config passed"
    @pytest.mark.execution_timeout(10)
    @unittest.skipIf(AirflowKubernetesScheduler is None, 'kubernetes python package is not installed')
    @mock.patch('airflow.executors.kubernetes_executor.AirflowKubernetesScheduler.run_pod_async')
    @mock.patch('airflow.executors.kubernetes_executor.get_kube_client')
    def test_pod_template_file_override_in_executor_config(self, mock_get_kube_client, mock_run_pod_async):