{"record":{"id":"53292dc8c4b95f2e","repo":"zhisheng17/flink-learning","slug":"could-not-get-the-rest-endpoint-of-clusterid","errorCode":null,"errorMessage":"Could not get the rest endpoint of ${clusterId}","messagePattern":"Could not get the rest endpoint of (.+?)","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"flink-learning-k8s/flink-k8s/src/main/java/org/apache/flink/kubernetes/KubernetesClusterDescriptor.java","lineNumber":98,"sourceCode":"\t\t\t\"ClusterId must be specified!\");\n\t}\n\n\t@Override\n\tpublic String getClusterDescription() {\n\t\treturn CLUSTER_DESCRIPTION;\n\t}\n\n\tprivate ClusterClientProvider<String> createClusterClientProvider(String clusterId) {\n\t\treturn () -> {\n\t\t\tfinal Configuration configuration = new Configuration(flinkConfig);\n\n\t\t\tfinal Optional<Endpoint> restEndpoint = client.getRestEndpoint(clusterId);\n\n\t\t\tif (restEndpoint.isPresent()) {\n\t\t\t\tconfiguration.setString(RestOptions.ADDRESS, restEndpoint.get().getAddress());\n\t\t\t\tconfiguration.setInteger(RestOptions.PORT, restEndpoint.get().getPort());\n\t\t\t} else {\n\t\t\t\tthrow new RuntimeException(\n\t\t\t\t\t\tnew ClusterRetrieveException(\n\t\t\t\t\t\t\t\t\"Could not get the rest endpoint of \" + clusterId));\n\t\t\t}\n\n\t\t\ttry {\n\t\t\t\t// Flink client will always use Kubernetes service to contact with jobmanager. So we have a pre-configured web\n\t\t\t\t// monitor address. Using StandaloneClientHAServices to create RestClusterClient is reasonable.\n\t\t\t\treturn new RestClusterClient<>(\n\t\t\t\t\tconfiguration,\n\t\t\t\t\tclusterId,\n\t\t\t\t\tnew StandaloneClientHAServices(getWebMonitorAddress(configuration)));\n\t\t\t} catch (Exception e) {\n\t\t\t\tclient.handleException(e);\n\t\t\t\tthrow new RuntimeException(new ClusterRetrieveException(\"Could not create the RestClusterClient.\", e));\n\t\t\t}\n\t\t};\n\t}\n","sourceCodeStart":80,"sourceCodeEnd":116,"githubUrl":"https://github.com/zhisheng17/flink-learning/blob/d731cee7618021be56d132cc925102ffff8d75e6/flink-learning-k8s/flink-k8s/src/main/java/org/apache/flink/kubernetes/KubernetesClusterDescriptor.java#L80-L116","documentation":"Thrown by KubernetesClusterDescriptor.createClusterClientProvider when the FlinkKubeClient returns no REST endpoint for the given clusterId, meaning the JobManager service exists but did not expose a usable REST endpoint. It wraps a ClusterRetrieveException inside a RuntimeException, aborting retrieval of a client for an existing cluster.","triggerScenarios":"Calling clusterClientProvider or deployClusterInternal after deployment when the Kubernetes REST service for clusterId was not found or had no endpoint addresses (e.g. service exists but JobManager pod not ready/no endpoints).","commonSituations":"Retrieving a cluster immediately after deploying before the JobManager pod becomes Ready; JobManager crashed after service creation; kube-proxy/DNS issues leaving the service with no endpoints; wrong clusterId pointing at a stale service.","solutions":["Wait for the JobManager pod and its REST service endpoints to become Ready before calling getRestEndpoint (kubectl get endpoints <clusterId>).","Verify the clusterId matches an actually running Flink cluster (kubectl get svc | grep <clusterId>).","Check JobManager pod logs/events for startup crashes and fix the underlying deployment failure.","Inspect Kubernetes service labels/selectors — if the selector does not match the JobManager pod the service will have no endpoints.","Increase timeout/retry logic around cluster retrieval in your launcher."],"exampleFix":"// before: retrieve immediately after deploy\nClusterClient<String> client = descriptor.deploySessionCluster(spec).getClusterClient();\n// after: wait for REST service endpoints first\nawait().atMost(2, MINUTES).until(() ->\n    kubeClient.getService(KubernetesService.ServiceType.REST_SERVICE, clusterId).isPresent()\n    && kubectlEndpointsReady(clusterId));\nClusterClient<String> client = provider.getClusterClient();","handlingStrategy":"retry","validationCode":"// verify REST service has endpoints before retrieval\nOptional<KubernetesService> svc = kubeClient.getService(KubernetesService.ServiceType.REST_SERVICE, clusterId);\nif (svc.isEmpty()) throw new IllegalStateException(\"REST service for \" + clusterId + \" missing\");","typeGuard":null,"tryCatchPattern":"try {\n    ClusterClient<String> client = provider.getClusterClient();\n} catch (RuntimeException e) {\n    if (e.getCause() instanceof ClusterRetrieveException) {\n        // back off and retry until JobManager endpoints are Ready\n    }\n}","preventionTips":["Wait for pod Ready + service endpoints before retrieving the client","Confirm clusterId matches a live cluster","Monitor JobManager startup logs","Add bounded retry with backoff around cluster retrieval"],"tags":["kubernetes","flink","rest-endpoint","cluster-retrieval"],"backgroundTag":"resource-not-found","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"}