[FLINK-38148] Support Flink v1.17 when listing job exceptions

Author: fijolekProjects

flink-kubernetes-operator/src/main/java/org/apache/flink/runtime/rest/messages/JobExceptionsInfoWithHistory.java

@@ -35,7 +35,10 @@
 import static org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonInclude.Include.NON_NULL;
 import static org.apache.flink.util.Preconditions.checkNotNull;
 
-/** Copied from Flink 2.0 to handle removed changes. */
+/**
+ * Copied from Flink 2.0 to handle removed changes, with small modifications like: FLINK-38148 that
+ * allows nullable `failureLabels` to support Flink v1.17
+ */
 public class JobExceptionsInfoWithHistory implements ResponseBody {
 
     public static final String FIELD_NAME_EXCEPTION_HISTORY = "exceptionHistory";
@@ -189,14 +192,15 @@ public ExceptionInfo(
                 @JsonProperty(FIELD_NAME_EXCEPTION_NAME) String exceptionName,
                 @JsonProperty(FIELD_NAME_EXCEPTION_STACKTRACE) String stacktrace,
                 @JsonProperty(FIELD_NAME_EXCEPTION_TIMESTAMP) long timestamp,
-                @JsonProperty(FIELD_NAME_FAILURE_LABELS) Map<String, String> failureLabels,
+                @JsonProperty(FIELD_NAME_FAILURE_LABELS) @Nullable
+                        Map<String, String> failureLabels,
                 @JsonProperty(FIELD_NAME_TASK_NAME) @Nullable String taskName,
                 @JsonProperty(FIELD_NAME_ENDPOINT) @Nullable String endpoint,
                 @JsonProperty(FIELD_NAME_TASK_MANAGER_ID) @Nullable String taskManagerId) {
             this.exceptionName = checkNotNull(exceptionName);
             this.stacktrace = checkNotNull(stacktrace);
             this.timestamp = timestamp;
-            this.failureLabels = checkNotNull(failureLabels);
+            this.failureLabels = failureLabels;
             this.taskName = taskName;
             this.endpoint = endpoint;
             this.taskManagerId = taskManagerId;

flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkServiceTest.java

@@ -59,6 +59,7 @@
 import org.apache.flink.runtime.rest.handler.async.AsynchronousOperationResult;
 import org.apache.flink.runtime.rest.handler.async.TriggerResponse;
 import org.apache.flink.runtime.rest.messages.DashboardConfiguration;
+import org.apache.flink.runtime.rest.messages.JobExceptionsInfoWithHistory;
 import org.apache.flink.runtime.rest.messages.MessageHeaders;
 import org.apache.flink.runtime.rest.messages.MessageParameters;
 import org.apache.flink.runtime.rest.messages.RequestBody;
@@ -1315,6 +1316,36 @@ public void testBlockingDeletionDeleteCallErrorHandling() {
         assertTrue(remaining.toMillis() < 1000);
     }
 
+    @Test
+    public void listingJobExceptionsIsCompatibleWihFlinkV1_17Test() throws Exception {
+        JobExceptionsInfoWithHistory exceptionInfoWithNullFailureLabels =
+                new JobExceptionsInfoWithHistory(
+                        new JobExceptionsInfoWithHistory.JobExceptionHistory(
+                                java.util.Collections.singletonList(
+                                        new JobExceptionsInfoWithHistory.RootExceptionInfo(
+                                                "org.apache.flink.util.FlinkExpectedException",
+                                                "The TaskExecutor is shutting down.",
+                                                1755998006623L,
+                                                null,
+                                                "Source: Events Generator Source (2/2) - execution #0",
+                                                "10.244.0.105:44401",
+                                                "basic-example-taskmanager-1-1",
+                                                java.util.Collections.emptyList())),
+                                false));
+
+        var flinkService =
+                getTestingService(
+                        (messageHeaders, messageParameters, requestBody) ->
+                                CompletableFuture.completedFuture(
+                                        exceptionInfoWithNullFailureLabels));
+        assertDoesNotThrow(
+                () ->
+                        flinkService.getJobExceptions(
+                                TestUtils.buildApplicationCluster(),
+                                new JobID(),
+                                new Configuration()));
+    }
+
     class TestingService extends AbstractFlinkService {
 
         RestClusterClient<String> clusterClient;