@Override
    public void stopAndCleanupCluster(String clusterId) {
        this.internalClient
                .apps()
                .deployments()
                .withName(KubernetesUtils.getDeploymentName(clusterId))
                .cascading(true)
                .delete();
    }
@Test
    void testStopAndCleanupCluster() throws Exception {
        this.flinkKubeClient.createJobManagerComponent(this.kubernetesJobManagerSpecification);

        final KubernetesPod kubernetesPod = buildKubernetesPod(TASKMANAGER_POD_NAME);
        this.flinkKubeClient.createTaskManagerPod(kubernetesPod).get();

        assertThat(
                        this.kubeClient
                                .apps()
                                .deployments()
                                .inNamespace(NAMESPACE)
                                .list()
                                .getItems()
                                .size())
                .isEqualTo(1);
        assertThat(this.kubeClient.configMaps().inNamespace(NAMESPACE).list().getItems().size())
                .isEqualTo(1);
        assertThat(this.kubeClient.services().inNamespace(NAMESPACE).list().getItems()).hasSize(2);
        assertThat(this.kubeClient.pods().inNamespace(NAMESPACE).list().getItems()).hasSize(1);

        this.flinkKubeClient.stopAndCleanupCluster(CLUSTER_ID);
        assertThat(this.kubeClient.apps().deployments().inNamespace(NAMESPACE).list().getItems())
                .isEmpty();
    }