@Override
    public void killCluster(String clusterId) throws FlinkException {
        try {
            client.stopAndCleanupCluster(clusterId);
        } catch (Exception e) {
            throw new FlinkException("Could not kill Kubernetes cluster " + clusterId);
        }
    }
@Test
    void testKillCluster() throws Exception {
        flinkConfig.set(DeploymentOptions.TARGET, KubernetesDeploymentTarget.SESSION.getName());
        deploySessionCluster();

        assertThat(kubeClient.services().list().getItems()).hasSize(2);

        descriptor.killCluster(CLUSTER_ID);

        // Mock kubernetes server do not delete the accompanying resources by gc.
        assertThat(kubeClient.apps().deployments().list().getItems()).isEmpty();
        assertThat(kubeClient.services().list().getItems()).hasSize(2);
        assertThat(kubeClient.configMaps().list().getItems()).hasSize(1);
    }