[FLINK-37730][Job Manager] Expose JM exception as K8s exceptions

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkOperatorConfiguration.java

@@ -77,6 +77,8 @@ public class FlinkOperatorConfiguration {
     DeletionPropagation deletionPropagation;
     boolean snapshotResourcesEnabled;
     Duration slowRequestThreshold;
+    int reportedExceptionEventsMaxCount;
+    int reportedExceptionEventsMaxStackTraceLength;
 
     public static FlinkOperatorConfiguration fromConfiguration(Configuration operatorConfig) {
         Duration reconcileInterval =
@@ -195,6 +197,12 @@ public static FlinkOperatorConfiguration fromConfiguration(Configuration operato
         Duration slowRequestThreshold =
                 operatorConfig.get(OPERATOR_KUBERNETES_SLOW_REQUEST_THRESHOLD);
 
+        int reportedExceptionEventsMaxCount =
+                operatorConfig.get(KubernetesOperatorConfigOptions.OPERATOR_EVENT_EXCEPTION_LIMIT);
+        int reportedExceptionEventsMaxStackTraceLength =
+                operatorConfig.get(
+                        KubernetesOperatorConfigOptions.OPERATOR_EVENT_EXCEPTION_STACKTRACE_LINES);
+
         return new FlinkOperatorConfiguration(
                 reconcileInterval,
                 reconcilerMaxParallelism,
@@ -224,7 +232,9 @@ public static FlinkOperatorConfiguration fromConfiguration(Configuration operato
                 getLeaderElectionConfig(operatorConfig),
                 deletionPropagation,
                 snapshotResourcesEnabled,
-                slowRequestThreshold);
+                slowRequestThreshold,
+                reportedExceptionEventsMaxCount,
+                reportedExceptionEventsMaxStackTraceLength);
     }
 
     private static GenericRetry getRetryConfig(Configuration conf) {

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java

@@ -654,4 +654,20 @@ public static String operatorConfigKey(String key) {
                     .defaultValue(Duration.ofMinutes(-1))
                     .withDescription(
                             "How often to retrieve Kubernetes cluster resource usage information. This information is used to avoid running out of cluster resources when scaling up resources. Negative values disable the feature.");
+
+    @Documentation.Section(SECTION_ADVANCED)
+    public static final ConfigOption<Integer> OPERATOR_EVENT_EXCEPTION_STACKTRACE_LINES =
+            operatorConfig("events.exceptions.stacktrace-lines")
+                    .intType()
+                    .defaultValue(5)
+                    .withDescription(
+                            "Maximum number of stack trace lines to include in exception-related Kubernetes event messages.");
+
+    @Documentation.Section(SECTION_ADVANCED)
+    public static final ConfigOption<Integer> OPERATOR_EVENT_EXCEPTION_LIMIT =
+            operatorConfig("events.exceptions.limit-per-reconciliation")
+                    .intType()
+                    .defaultValue(10)
+                    .withDescription(
+                            "Maximum number of exception-related Kubernetes events emitted per reconciliation cycle.");
 }

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentContext.java

@@ -25,6 +25,7 @@
 import org.apache.flink.kubernetes.operator.api.spec.KubernetesDeploymentMode;
 import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
 import org.apache.flink.kubernetes.operator.metrics.KubernetesResourceMetricGroup;
+import org.apache.flink.kubernetes.operator.service.FlinkResourceContextFactory;
 import org.apache.flink.kubernetes.operator.service.FlinkService;
 
 import io.javaoperatorsdk.operator.api.reconciler.Context;
@@ -39,8 +40,15 @@ public FlinkDeploymentContext(
             Context<?> josdkContext,
             KubernetesResourceMetricGroup resourceMetricGroup,
             FlinkConfigManager configManager,
-            Function<FlinkResourceContext<?>, FlinkService> flinkServiceFactory) {
-        super(resource, josdkContext, resourceMetricGroup, configManager, flinkServiceFactory);
+            Function<FlinkResourceContext<?>, FlinkService> flinkServiceFactory,
+            FlinkResourceContextFactory.ExceptionCacheEntry exceptionCacheEntry) {
+        super(
+                resource,
+                josdkContext,
+                resourceMetricGroup,
+                configManager,
+                flinkServiceFactory,
+                exceptionCacheEntry);
     }
 
     @Override

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkResourceContext.java

@@ -30,6 +30,7 @@
 import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
 import org.apache.flink.kubernetes.operator.config.FlinkOperatorConfiguration;
 import org.apache.flink.kubernetes.operator.metrics.KubernetesResourceMetricGroup;
+import org.apache.flink.kubernetes.operator.service.FlinkResourceContextFactory;
 import org.apache.flink.kubernetes.operator.service.FlinkService;
 
 import io.fabric8.kubernetes.client.KubernetesClient;
@@ -51,6 +52,8 @@ public abstract class FlinkResourceContext<CR extends AbstractFlinkResource<?, ?
     protected final FlinkConfigManager configManager;
     private final Function<FlinkResourceContext<?>, FlinkService> flinkServiceFactory;
 
+    @Getter private final FlinkResourceContextFactory.ExceptionCacheEntry exceptionCacheEntry;
+
     private FlinkOperatorConfiguration operatorConfig;
     private Configuration observeConfig;
     private FlinkService flinkService;

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkSessionJobContext.java

@@ -26,6 +26,7 @@
 import org.apache.flink.kubernetes.operator.api.spec.KubernetesDeploymentMode;
 import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
 import org.apache.flink.kubernetes.operator.metrics.KubernetesResourceMetricGroup;
+import org.apache.flink.kubernetes.operator.service.FlinkResourceContextFactory;
 import org.apache.flink.kubernetes.operator.service.FlinkService;
 
 import io.javaoperatorsdk.operator.api.reconciler.Context;
@@ -44,8 +45,15 @@ public FlinkSessionJobContext(
             Context<?> josdkContext,
             KubernetesResourceMetricGroup resourceMetricGroup,
             FlinkConfigManager configManager,
-            Function<FlinkResourceContext<?>, FlinkService> flinkServiceFactory) {
-        super(resource, josdkContext, resourceMetricGroup, configManager, flinkServiceFactory);
+            Function<FlinkResourceContext<?>, FlinkService> flinkServiceFactory,
+            FlinkResourceContextFactory.ExceptionCacheEntry exceptionCacheEntry) {
+        super(
+                resource,
+                josdkContext,
+                resourceMetricGroup,
+                configManager,
+                flinkServiceFactory,
+                exceptionCacheEntry);
     }
 
     @Override

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/JobStatusObserver.java

@@ -19,6 +19,7 @@
 
 import org.apache.flink.api.common.JobID;
 import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.autoscaler.utils.DateTimeUtils;
 import org.apache.flink.kubernetes.operator.api.AbstractFlinkResource;
 import org.apache.flink.kubernetes.operator.api.FlinkSessionJob;
 import org.apache.flink.kubernetes.operator.api.spec.JobState;
@@ -29,10 +30,15 @@
 import org.apache.flink.kubernetes.operator.utils.EventRecorder;
 import org.apache.flink.kubernetes.operator.utils.ExceptionUtils;
 import org.apache.flink.runtime.client.JobStatusMessage;
+import org.apache.flink.runtime.rest.messages.JobExceptionsInfoWithHistory;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.time.Instant;
+import java.time.ZoneId;
+import java.util.HashMap;
+import java.util.Map;
 import java.util.concurrent.TimeoutException;
 
 import static org.apache.flink.kubernetes.operator.utils.FlinkResourceExceptionUtils.updateFlinkResourceException;
@@ -69,17 +75,23 @@ public boolean observe(FlinkResourceContext<R> ctx) {
         var jobStatus = resource.getStatus().getJobStatus();
         LOG.debug("Observing job status");
         var previousJobStatus = jobStatus.getState();
-
+        var jobId = jobStatus.getJobId();
         try {
             var newJobStatusOpt =
                     ctx.getFlinkService()
-                            .getJobStatus(
-                                    ctx.getObserveConfig(),
-                                    JobID.fromHexString(jobStatus.getJobId()));
+                            .getJobStatus(ctx.getObserveConfig(), JobID.fromHexString(jobId));
 
             if (newJobStatusOpt.isPresent()) {
-                updateJobStatus(ctx, newJobStatusOpt.get());
+                var newJobStatus = newJobStatusOpt.get();
+                updateJobStatus(ctx, newJobStatus);
                 ReconciliationUtils.checkAndUpdateStableSpec(resource.getStatus());
+                // now check if the job is in a terminal state. This might not be need as we have
+                // already
+                // verified that the REST API server is available
+                // now check if the job is NOT in a terminal state
+                if (!newJobStatus.getJobState().isGloballyTerminalState()) {
+                    observeJobManagerExceptions(ctx);
+                }
                 return true;
             } else {
                 onTargetJobNotFound(ctx);
@@ -95,6 +107,143 @@ public boolean observe(FlinkResourceContext<R> ctx) {
         return false;
     }
 
+    /**
+     * Observe the exceptions raised in the job manager and take appropriate action.
+     *
+     * @param ctx the context with which the operation is executed
+     */
+    protected void observeJobManagerExceptions(FlinkResourceContext<R> ctx) {
+        var resource = ctx.getResource();
+        var operatorConfig = ctx.getOperatorConfig();
+        var jobStatus = resource.getStatus().getJobStatus();
+
+        try {
+            var jobId = JobID.fromHexString(jobStatus.getJobId());
+            // TODO: Ideally the best way to restrict the number of events is to use the query param
+            // `maxExceptions`
+            //  but the JobExceptionsMessageParameters does not expose the parameters and nor does
+            // it have setters.
+            var history =
+                    ctx.getFlinkService()
+                            .getJobExceptions(
+                                    resource, jobId, ctx.getDeployConfig(resource.getSpec()));
+
+            if (history == null || history.getExceptionHistory() == null) {
+                return;
+            }
+
+            var exceptionHistory = history.getExceptionHistory();
+            var exceptions = exceptionHistory.getEntries();
+            if (exceptions.isEmpty()) {
+                return;
+            }
+
+            if (exceptionHistory.isTruncated()) {
+                LOG.warn(
+                        "Job exception history is truncated for jobId '{}'. Some exceptions may be missing.",
+                        jobId);
+            }
+
+            String currentJobId = jobStatus.getJobId();
+            Instant lastRecorded = null; // first reconciliation
+
+            var cacheEntry = ctx.getExceptionCacheEntry();
+            // a cache entry is created should always be present. The timestamp for the first
+            // reconciliation would be
+            // when the job was created. This check is still necessary because even though there
+            // might be an entry,
+            // the jobId could have changed since the job was first created.
+            if (cacheEntry.getJobId().equals(currentJobId)) {
+                lastRecorded = Instant.ofEpochMilli(cacheEntry.getLastTimestamp());
+            }
+
+            Instant now = Instant.now();
+            int maxEvents = operatorConfig.getReportedExceptionEventsMaxCount();
+            int maxStackTraceLines = operatorConfig.getReportedExceptionEventsMaxStackTraceLength();
+
+            int count = 0;
+            for (var exception : exceptions) {
+                Instant exceptionTime = Instant.ofEpochMilli(exception.getTimestamp());
+                if (lastRecorded != null && exceptionTime.isBefore(lastRecorded)) {
+                    continue;
+                }
+
+                emitJobManagerExceptionEvent(ctx, exception, exceptionTime, maxStackTraceLines);
+                if (++count >= maxEvents) {
+                    break;
+                }
+            }
+            ctx.getExceptionCacheEntry().setJobId(currentJobId);
+            ctx.getExceptionCacheEntry().setLastTimestamp(now.toEpochMilli());
+        } catch (Exception e) {
+            LOG.warn("Failed to fetch JobManager exception info.", e);
+        }
+    }
+
+    private void emitJobManagerExceptionEvent(
+            FlinkResourceContext<R> ctx,
+            JobExceptionsInfoWithHistory.RootExceptionInfo exception,
+            Instant exceptionTime,
+            int maxStackTraceLines) {
+
+        String exceptionName = exception.getExceptionName();
+        if (exceptionName == null || exceptionName.isBlank()) {
+            return;
+        }
+
+        Map<String, String> annotations = new HashMap<>();
+        annotations.put(
+                "event-time-readable",
+                DateTimeUtils.readable(exceptionTime, ZoneId.systemDefault()));
+        annotations.put("event-timestamp-millis", String.valueOf(exceptionTime.toEpochMilli()));
+
+        if (exception.getTaskName() != null) {
+            annotations.put("task-name", exception.getTaskName());
+        }
+        if (exception.getEndpoint() != null) {
+            annotations.put("endpoint", exception.getEndpoint());
+        }
+        if (exception.getTaskManagerId() != null) {
+            annotations.put("tm-id", exception.getTaskManagerId());
+        }
+
+        if (exception.getFailureLabels() != null) {
+            exception
+                    .getFailureLabels()
+                    .forEach((k, v) -> annotations.put("failure-label-" + k, v));
+        }
+
+        StringBuilder eventMessage = new StringBuilder(exceptionName);
+        String stacktrace = exception.getStacktrace();
+        if (stacktrace != null && !stacktrace.isBlank()) {
+            String[] lines = stacktrace.split("\n");
+            eventMessage.append("\n\nStacktrace (truncated):\n");
+            for (int i = 0; i < Math.min(maxStackTraceLines, lines.length); i++) {
+                eventMessage.append(lines[i]).append("\n");
+            }
+            if (lines.length > maxStackTraceLines) {
+                eventMessage
+                        .append("... (")
+                        .append(lines.length - maxStackTraceLines)
+                        .append(" more lines)");
+            }
+        }
+
+        String keyMessage =
+                exceptionName.length() > 128 ? exceptionName.substring(0, 128) : exceptionName;
+
+        eventRecorder.triggerEventOnceWithAnnotations(
+                exceptionTime.toEpochMilli(),
+                ctx.getResource(),
+                EventRecorder.Type.Warning,
+                EventRecorder.Reason.JobException,
+                eventMessage.toString().trim(),
+                EventRecorder.Component.JobManagerDeployment,
+                "jobmanager-exception-" + keyMessage.hashCode(),
+                ctx.getKubernetesClient(),
+                annotations);
+    }
+
     /**
      * Callback when no matching target job was found on a cluster where jobs were found.
      *

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

@@ -31,6 +31,7 @@
 import org.apache.flink.core.execution.RestoreMode;
 import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions;
 import org.apache.flink.kubernetes.kubeclient.decorators.ExternalServiceDecorator;
+import org.apache.flink.kubernetes.operator.api.AbstractFlinkResource;
 import org.apache.flink.kubernetes.operator.api.FlinkDeployment;
 import org.apache.flink.kubernetes.operator.api.FlinkSessionJob;
 import org.apache.flink.kubernetes.operator.api.spec.FlinkSessionJobSpec;
@@ -65,6 +66,8 @@
 import org.apache.flink.runtime.rest.messages.DashboardConfiguration;
 import org.apache.flink.runtime.rest.messages.EmptyMessageParameters;
 import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.JobExceptionsHeaders;
+import org.apache.flink.runtime.rest.messages.JobExceptionsInfoWithHistory;
 import org.apache.flink.runtime.rest.messages.JobsOverviewHeaders;
 import org.apache.flink.runtime.rest.messages.TriggerId;
 import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointIdPathParameter;
@@ -77,6 +80,7 @@
 import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerRequestBody;
 import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointingStatistics;
 import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointingStatisticsHeaders;
+import org.apache.flink.runtime.rest.messages.job.JobExceptionsMessageParameters;
 import org.apache.flink.runtime.rest.messages.job.metrics.JobMetricsHeaders;
 import org.apache.flink.runtime.rest.messages.job.savepoints.SavepointDisposalRequest;
 import org.apache.flink.runtime.rest.messages.job.savepoints.SavepointDisposalTriggerHeaders;
@@ -845,6 +849,37 @@ public RestClusterClient<String> getClusterClient(Configuration conf) throws Exc
                 (c, e) -> new StandaloneClientHAServices(restServerAddress));
     }
 
+    @Override
+    public JobExceptionsInfoWithHistory getJobExceptions(
+            AbstractFlinkResource resource, JobID jobId, Configuration deployConfig) {
+        JobExceptionsHeaders jobExceptionsHeaders = JobExceptionsHeaders.getInstance();
+        int port = deployConfig.getInteger(RestOptions.PORT);
+        String host =
+                ObjectUtils.firstNonNull(
+                        operatorConfig.getFlinkServiceHostOverride(),
+                        ExternalServiceDecorator.getNamespacedExternalServiceName(
+                                resource.getMetadata().getName(),
+                                resource.getMetadata().getNamespace()));
+        JobExceptionsMessageParameters params = new JobExceptionsMessageParameters();
+        params.jobPathParameter.resolve(jobId);
+        try (var restClient = getRestClient(deployConfig)) {
+            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;
+        }
+    }
+
     @VisibleForTesting
     protected void runJar(
             JobSpec job,