Add fix for host:port for Session Job

Author: vsantwana

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java

@@ -851,32 +851,74 @@ public RestClusterClient<String> getClusterClient(Configuration conf) throws Exc
 
     @Override
     public JobExceptionsInfoWithHistory getJobExceptions(
-            AbstractFlinkResource resource, JobID jobId, Configuration observeConfig) {
+            AbstractFlinkResource resource, JobID jobId, Configuration observeConfig)
+            throws IOException {
         JobExceptionsHeaders jobExceptionsHeaders = JobExceptionsHeaders.getInstance();
+        // port is specified at the same place for both session and application jobs
         int port = observeConfig.getInteger(RestOptions.PORT);
-        String host =
-                ObjectUtils.firstNonNull(
-                        operatorConfig.getFlinkServiceHostOverride(),
-                        ExternalServiceDecorator.getNamespacedExternalServiceName(
-                                resource.getMetadata().getName(),
-                                resource.getMetadata().getNamespace()));
+        String host;
+        // In case of session job, there is a single rest service for all the jobs, while in the
+        // application mode,
+        // the rest service is per job.
+
+        Configuration operatorRestConf = observeConfig;
+        if (SecurityOptions.isRestSSLEnabled(observeConfig)) {
+            operatorRestConf = getOperatorRestConfig(observeConfig);
+        }
+
+        if (resource instanceof FlinkSessionJob) {
+            // For session jobs, we need to use the deployment name from the spec
+            FlinkSessionJob sessionJob = (FlinkSessionJob) resource;
+            final String clusterId = observeConfig.get(KubernetesConfigOptions.CLUSTER_ID);
+            final String namespace = observeConfig.get(KubernetesConfigOptions.NAMESPACE);
+            host =
+                    ObjectUtils.firstNonNull(
+                            operatorConfig.getFlinkServiceHostOverride(),
+                            ExternalServiceDecorator.getNamespacedExternalServiceName(
+                                    clusterId, namespace));
+        } else {
+            // For application jobs (FlinkDeployment), use the resource name directly
+            FlinkDeployment deployment = (FlinkDeployment) resource;
+            host =
+                    ObjectUtils.firstNonNull(
+                            operatorConfig.getFlinkServiceHostOverride(),
+                            ExternalServiceDecorator.getNamespacedExternalServiceName(
+                                    deployment.getMetadata().getName(),
+                                    resource.getMetadata().getNamespace()));
+        }
+
         JobExceptionsMessageParameters params = new JobExceptionsMessageParameters();
         params.jobPathParameter.resolve(jobId);
-        try (var restClient = getRestClient(observeConfig)) {
-            return restClient
-                    .sendRequest(
-                            host,
-                            port,
-                            jobExceptionsHeaders,
-                            params,
-                            EmptyRequestBody.getInstance())
-                    .get(operatorConfig.getFlinkClientTimeout().toSeconds(), TimeUnit.SECONDS);
-        } catch (Exception e) {
-            LOG.warn(
-                    String.format(
-                            "Failed to fetch job exceptions from REST API for jobId %s", jobId),
-                    e);
-            return null;
+
+        if (resource instanceof FlinkSessionJob) {
+            try (var clusterClient = getClusterClient(observeConfig)) {
+                return clusterClient
+                        .sendRequest(jobExceptionsHeaders, params, EmptyRequestBody.getInstance())
+                        .get(operatorConfig.getFlinkClientTimeout().toSeconds(), TimeUnit.SECONDS);
+            } catch (Exception e) {
+                LOG.warn(
+                        String.format(
+                                "Failed to fetch job exceptions from REST API for jobId %s", jobId),
+                        e);
+                return null;
+            }
+        } else {
+            try (var restClient = getRestClient(observeConfig)) {
+                return restClient
+                        .sendRequest(
+                                host,
+                                port,
+                                jobExceptionsHeaders,
+                                params,
+                                EmptyRequestBody.getInstance())
+                        .get(operatorConfig.getFlinkClientTimeout().toSeconds(), TimeUnit.SECONDS);
+            } catch (Exception e) {
+                LOG.warn(
+                        String.format(
+                                "Failed to fetch job exceptions from REST API for jobId %s", jobId),
+                        e);
+                return null;
+            }
         }
     }