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.start()
        test_time = timezone.utcnow()
        key = ('dag_id', 'task_id', test_time, 'try_number2')
        executor._change_state(key, State.SUCCESS, 'pod_id', 'default')
        assert executor.event_buffer[key][0] == State.SUCCESS
        mock_delete_pod.assert_called_once_with('pod_id', 'default')
    @mock.patch('airflow.executors.kubernetes_executor.KubernetesJobWatcher')
    @mock.patch('airflow.executors.kubernetes_executor.get_kube_client')
    @mock.patch('airflow.executors.kubernetes_executor.AirflowKubernetesScheduler.delete_pod')
    def test_change_state_failed_no_deletion(
        self, mock_delete_pod, mock_get_kube_client, mock_kubernetes_job_watcher
    ):
        executor = self.kubernetes_executor
        executor.kube_config.delete_worker_pods = False
        executor.kube_config.delete_worker_pods_on_failure = False
        executor.start()
        test_time = timezone.utcnow()
        key = ('dag_id', 'task_id', test_time, 'try_number3')
        executor._change_state(key, State.FAILED, 'pod_id', 'default')
        assert executor.event_buffer[key][0] == State.FAILED
        mock_delete_pod.assert_not_called()
    
    @mock.patch('airflow.executors.kubernetes_executor.KubernetesJobWatcher')
    @mock.patch('airflow.executors.kubernetes_executor.get_kube_client')
    @mock.patch('airflow.executors.kubernetes_executor.AirflowKubernetesScheduler.delete_pod')
    def test_change_state_skip_pod_deletion(