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
'try_number': '1',
                        },
                    ),
                    spec=k8s.V1PodSpec(
                        containers=[
                            k8s.V1Container(
                                name="base",
                                image="airflow:3.6",
                                args=['airflow', 'tasks', 'run', 'true', 'some_parameter'],
                                env=[k8s.V1EnvVar(name='AIRFLOW_IS_K8S_EXECUTOR_POD', value='True')],
                            )
                        ],
                        image_pull_secrets=[k8s.V1LocalObjectReference(name='airflow-registry')],
                        scheduler_name='default-scheduler',
                        security_context=k8s.V1PodSecurityContext(fs_group=50000, run_as_user=50000),
                    ),
                )
            )
    @mock.patch('airflow.executors.kubernetes_executor.KubernetesJobWatcher')
    @mock.patch('airflow.executors.kubernetes_executor.get_kube_client')
    def test_change_state_running(self, mock_get_kube_client, mock_kubernetes_job_watcher):
        executor = self.kubernetes_executor
        executor.start()
        key = ('dag_id', 'task_id', 'ex_time', 'try_number1')
        executor._change_state(key, State.RUNNING, 'pod_id', 'default')
        assert executor.event_buffer[key][0] == State.RUNNING
    @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_success(self, mock_delete_pod, mock_get_kube_client, mock_kubernetes_job_watcher):