{"record":{"id":"aec6014a525f9cd6","repo":"zhisheng17/flink-learning","slug":"the-flink-cluster-clusterid-already-exists","errorCode":null,"errorMessage":"The Flink cluster ${clusterId} already exists.","messagePattern":"The Flink cluster (.+?) already exists\\.","errorType":"exception","errorClass":"ClusterDeploymentException","httpStatus":null,"severity":"error","filePath":"flink-learning-k8s/flink-k8s/src/main/java/org/apache/flink/kubernetes/KubernetesClusterDescriptor.java","lineNumber":168,"sourceCode":"\t\t\tclusterSpecification,\n\t\t\tfalse);\n\n\t\ttry (ClusterClient<String> clusterClient = clusterClientProvider.getClusterClient()) {\n\t\t\tLOG.info(\n\t\t\t\t\"Create flink session cluster {} successfully, JobManager Web Interface: {}\",\n\t\t\t\tclusterId,\n\t\t\t\tclusterClient.getWebInterfaceURL());\n\t\t}\n\t\treturn clusterClientProvider;\n\t}\n\n\t@Override\n\tpublic ClusterClientProvider<String> deployApplicationCluster(\n\t\t\tfinal ClusterSpecification clusterSpecification,\n\t\t\tfinal ApplicationConfiguration applicationConfiguration) throws ClusterDeploymentException {\n\t\t//todo：k8s application mode\n\t\tif (client.getService(KubernetesService.ServiceType.REST_SERVICE, clusterId).isPresent()) {\n\t\t\tthrow new ClusterDeploymentException(\"The Flink cluster \" + clusterId + \" already exists.\");\n\t\t}\n\n\t\tcheckNotNull(clusterSpecification);\n\t\tcheckNotNull(applicationConfiguration);\n\n\t\tfinal KubernetesDeploymentTarget deploymentTarget = KubernetesDeploymentTarget.fromConfig(flinkConfig);\n\t\tif (KubernetesDeploymentTarget.APPLICATION != deploymentTarget) {\n\t\t\tthrow new ClusterDeploymentException(\n\t\t\t\t\"Couldn't deploy Kubernetes Application Cluster.\" +\n\t\t\t\t\t\" Expected deployment.target=\" + KubernetesDeploymentTarget.APPLICATION.getName() +\n\t\t\t\t\t\" but actual one was \\\"\" + deploymentTarget + \"\\\"\");\n\t\t}\n\n\t\tapplicationConfiguration.applyToConfiguration(flinkConfig);\n\n\t\t// No need to do pipelineJars validation if it is a PyFlink job.\n\t\tif (!(PackagedProgramUtils.isPython(applicationConfiguration.getApplicationClassName()) ||\n\t\t\tPackagedProgramUtils.isPython(applicationConfiguration.getProgramArguments()))) {","sourceCodeStart":150,"sourceCodeEnd":186,"githubUrl":"https://github.com/zhisheng17/flink-learning/blob/d731cee7618021be56d132cc925102ffff8d75e6/flink-learning-k8s/flink-k8s/src/main/java/org/apache/flink/kubernetes/KubernetesClusterDescriptor.java#L150-L186","documentation":"Thrown by deployApplicationCluster when a Kubernetes service for the target clusterId already exists, indicating a Flink cluster with that name is already deployed. It is a guard against deploying duplicate clusters and colliding Kubernetes resources.","triggerScenarios":"Calling deployApplicationCluster while a REST service with the same clusterId is present in the namespace (previous deployment not cleaned up, or an intentional name reuse).","commonSituations":"Re-running a job submission script without killing the old cluster; leftover resources from a crashed deployment; two concurrent submissions using the same cluster-id in one namespace; switching from session to application mode with a reused cluster-id.","solutions":["Pick a unique cluster-id (e.g. append a timestamp/UUID) for the new deployment.","Kill the existing cluster first: flink kill / KubernetesClusterDescriptor.killCluster(clusterId), or kubectl delete deployments/svc with the cluster's labels.","Check for residual resources with kubectl get svc,deploy | grep <clusterId> and clean them up manually.","Ensure only one submission process runs per cluster-id (avoid concurrent submits)."],"exampleFix":"// before\nString clusterId = \"my-flink-cluster\";\ndescriptor.deployApplicationCluster(spec, appConfig);\n// after\nString clusterId = \"my-flink-cluster-\" + UUID.randomUUID().toString().substring(0, 8);\ndescriptor.deployApplicationCluster(spec, appConfig);","handlingStrategy":"validation","validationCode":"boolean exists = kubeClient.getService(KubernetesService.ServiceType.REST_SERVICE, clusterId).isPresent();\nif (exists) { /* delete old cluster or pick a new clusterId */ }","typeGuard":null,"tryCatchPattern":"try {\n    descriptor.deployApplicationCluster(spec, appConfig);\n} catch (ClusterDeploymentException e) {\n    if (e.getMessage().contains(\"already exists\")) {\n        descriptor.killCluster(clusterId); // then redeploy\n    }\n}","preventionTips":["Use unique cluster-ids (append UUID/timestamp)","Clean up clusters in shutdown/finally blocks","Avoid concurrent submissions with the same cluster-id","Check for residual svc/deploy resources before deploying"],"tags":["kubernetes","flink","duplicate-resource","cluster-deployment"],"backgroundTag":"file-already-exists","analyzedSha":"d731cee7618021be56d132cc925102ffff8d75e6","analyzedAt":"2026-09-06T05:35:08.496Z","contentChangedAt":"2026-09-06T05:35:08.496Z","schemaVersion":2},"datasetVersion":"2026-09-14T00:17:10.932Z"}