@Override
    public ClusterClientProvider<String> deploySessionCluster(
            ClusterSpecification clusterSpecification) throws ClusterDeploymentException {
        final ClusterClientProvider<String> clusterClientProvider =
                deployClusterInternal(
                        KubernetesSessionClusterEntrypoint.class.getName(),
                        clusterSpecification,
                        false);

        try (ClusterClient<String> clusterClient = clusterClientProvider.getClusterClient()) {
            LOG.info(
                    "Create flink session cluster {} successfully, JobManager Web Interface: {}",
                    clusterId,
                    clusterClient.getWebInterfaceURL());
        }
        return clusterClientProvider;
    }
@Test
    void testDeploySessionCluster() throws Exception {
        flinkConfig.set(DeploymentOptions.TARGET, KubernetesDeploymentTarget.SESSION.getName());
        final ClusterClient<String> clusterClient = deploySessionCluster().getClusterClient();
        checkClusterClient(clusterClient);
        checkUpdatedConfigAndResourceSetting();
        clusterClient.close();
    }