[FLINK-25096] Fixes empty exception history for JobInitializationExce…

…ption

flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/ExecutionGraphInfo.java

@@ -38,7 +38,13 @@ public class ExecutionGraphInfo implements Serializable {
     private final Iterable<RootExceptionHistoryEntry> exceptionHistory;
 
     public ExecutionGraphInfo(ArchivedExecutionGraph executionGraph) {
-        this(executionGraph, Collections.emptyList());
+        this(
+                executionGraph,
+                executionGraph.getFailureInfo() != null
+                        ? Collections.singleton(
+                                RootExceptionHistoryEntry.fromGlobalFailure(
+                                        executionGraph.getFailureInfo()))
+                        : Collections.emptyList());
     }
 
     public ExecutionGraphInfo(

flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/exceptionhistory/RootExceptionHistoryEntry.java

@@ -19,11 +19,14 @@
 package org.apache.flink.runtime.scheduler.exceptionhistory;
 
 import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.runtime.executiongraph.ErrorInfo;
 import org.apache.flink.runtime.executiongraph.Execution;
 import org.apache.flink.runtime.taskmanager.TaskManagerLocation;
+import org.apache.flink.util.Preconditions;
 
 import javax.annotation.Nullable;
 
+import java.util.Collections;
 import java.util.Set;
 import java.util.stream.Collectors;
 import java.util.stream.StreamSupport;
@@ -84,6 +87,24 @@ public static RootExceptionHistoryEntry fromGlobalFailure(
         return createRootExceptionHistoryEntry(cause, timestamp, null, null, executions);
     }
 
+    /**
+     * Creates a {@code RootExceptionHistoryEntry} based on the passed {@link ErrorInfo}. No
+     * concurrent failures will be added.
+     *
+     * @param errorInfo The failure information that shall be used to initialize the {@code
+     *     RootExceptionHistoryEntry}.
+     * @return The {@code RootExceptionHistoryEntry} instance.
+     * @throws NullPointerException if {@code errorInfo} is {@code null} or the passed info does not
+     *     contain a {@code Throwable}.
+     * @throws IllegalArgumentException if the passed {@code timestamp} is not bigger than {@code
+     *     0}.
+     */
+    public static RootExceptionHistoryEntry fromGlobalFailure(ErrorInfo errorInfo) {
+        Preconditions.checkNotNull(errorInfo);
+        return fromGlobalFailure(
+                errorInfo.getException(), errorInfo.getTimestamp(), Collections.emptyList());
+    }
+
     private static RootExceptionHistoryEntry createRootExceptionHistoryEntry(
             Throwable cause,
             long timestamp,

flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/DefaultJobMasterServiceProcessTest.java

@@ -22,25 +22,32 @@
 import org.apache.flink.api.common.JobStatus;
 import org.apache.flink.runtime.client.JobInitializationException;
 import org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph;
+import org.apache.flink.runtime.executiongraph.ErrorInfo;
 import org.apache.flink.runtime.jobmaster.factories.TestingJobMasterServiceFactory;
 import org.apache.flink.runtime.jobmaster.utils.TestingJobMasterGateway;
 import org.apache.flink.runtime.jobmaster.utils.TestingJobMasterGatewayBuilder;
 import org.apache.flink.runtime.rest.handler.legacy.utils.ArchivedExecutionGraphBuilder;
 import org.apache.flink.runtime.scheduler.ExecutionGraphInfo;
+import org.apache.flink.runtime.scheduler.exceptionhistory.RootExceptionHistoryEntry;
+import org.apache.flink.util.ExceptionUtils;
 import org.apache.flink.util.TestLogger;
 
 import org.junit.Test;
 
 import java.time.Duration;
+import java.util.Iterator;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
 import java.util.function.Function;
 
 import static org.apache.flink.core.testutils.FlinkMatchers.containsCause;
 import static org.apache.flink.core.testutils.FlinkMatchers.containsMessage;
 import static org.apache.flink.core.testutils.FlinkMatchers.futureWillCompleteExceptionally;
 import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
@@ -56,7 +63,8 @@ public class DefaultJobMasterServiceProcessTest extends TestLogger {
                                     jobId, "test", JobStatus.FAILED, throwable, null, 1337));
 
     @Test
-    public void testInitializationFailureCompletesResultFuture() {
+    public void testInitializationFailureCompletesResultFuture()
+            throws ExecutionException, InterruptedException {
         final CompletableFuture<JobMasterService> jobMasterServiceFuture =
                 new CompletableFuture<>();
         DefaultJobMasterServiceProcess serviceProcess = createTestInstance(jobMasterServiceFuture);
@@ -70,6 +78,69 @@ public void testInitializationFailureCompletesResultFuture() {
         assertThat(initializationFailure, containsCause(originalCause));
     }
 
+    @Test
+    public void testInitializationFailureSetsFailureInfoProperly()
+            throws ExecutionException, InterruptedException {
+        final CompletableFuture<JobMasterService> jobMasterServiceFuture =
+                new CompletableFuture<>();
+        DefaultJobMasterServiceProcess serviceProcess = createTestInstance(jobMasterServiceFuture);
+        final RuntimeException originalCause = new RuntimeException("Expected RuntimeException");
+
+        long beforeFailureTimestamp = System.currentTimeMillis();
+        jobMasterServiceFuture.completeExceptionally(originalCause);
+        long afterFailureTimestamp = System.currentTimeMillis();
+
+        final JobManagerRunnerResult result = serviceProcess.getResultFuture().get();
+        final ErrorInfo executionGraphFailure =
+                result.getExecutionGraphInfo().getArchivedExecutionGraph().getFailureInfo();
+
+        assertNotNull(executionGraphFailure);
+        final Throwable actualException =
+                executionGraphFailure
+                        .getException()
+                        .deserializeError(Thread.currentThread().getContextClassLoader());
+        assertEquals(JobInitializationException.class, actualException.getClass());
+        assertTrue(
+                ExceptionUtils.findThrowable(actualException, t -> t.equals(originalCause))
+                        .isPresent());
+
+        assertTrue(executionGraphFailure.getTimestamp() >= beforeFailureTimestamp);
+        assertTrue(executionGraphFailure.getTimestamp() <= afterFailureTimestamp);
+    }
+
+    @Test
+    public void testInitializationFailureSetsExceptionHistoryProperly()
+            throws ExecutionException, InterruptedException {
+        final CompletableFuture<JobMasterService> jobMasterServiceFuture =
+                new CompletableFuture<>();
+        DefaultJobMasterServiceProcess serviceProcess = createTestInstance(jobMasterServiceFuture);
+        final RuntimeException originalCause = new RuntimeException("Expected RuntimeException");
+
+        long beforeFailureTimestamp = System.currentTimeMillis();
+        jobMasterServiceFuture.completeExceptionally(originalCause);
+        long afterFailureTimestamp = System.currentTimeMillis();
+
+        final JobManagerRunnerResult result = serviceProcess.getResultFuture().get();
+        final Iterator<RootExceptionHistoryEntry> exceptionHistoryEntryIterator =
+                result.getExecutionGraphInfo().getExceptionHistory().iterator();
+
+        assertTrue(exceptionHistoryEntryIterator.hasNext());
+        final RootExceptionHistoryEntry entry = exceptionHistoryEntryIterator.next();
+        assertFalse(exceptionHistoryEntryIterator.hasNext());
+
+        final Throwable actualException =
+                entry.getException()
+                        .deserializeError(Thread.currentThread().getContextClassLoader());
+        assertEquals(JobInitializationException.class, actualException.getClass());
+        assertTrue(
+                ExceptionUtils.findThrowable(actualException, t -> t.equals(originalCause))
+                        .isPresent());
+
+        assertTrue(entry.getTimestamp() >= beforeFailureTimestamp);
+        assertTrue(entry.getTimestamp() <= afterFailureTimestamp);
+        assertTrue(entry.isGlobal());
+    }
+
     @Test
     public void testCloseAfterInitializationFailure() throws Exception {
         final CompletableFuture<JobMasterService> jobMasterServiceFuture =

flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/ExecutionGraphInfoTest.java

@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.scheduler;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph;
+import org.apache.flink.runtime.executiongraph.ErrorInfo;
+import org.apache.flink.runtime.scheduler.exceptionhistory.RootExceptionHistoryEntry;
+
+import org.junit.Test;
+
+import java.util.Iterator;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+/** {@code ExecutionGraphInfoTest} tests the proper initialization of {@link ExecutionGraphInfo}. */
+public class ExecutionGraphInfoTest {
+
+    @Test
+    public void testExecutionGraphHistoryBeingDerivedFromFailedExecutionGraph() {
+        final ArchivedExecutionGraph executionGraph =
+                ArchivedExecutionGraph.createFromInitializingJob(
+                        new JobID(),
+                        "test job name",
+                        JobStatus.FAILED,
+                        new RuntimeException("Expected RuntimeException"),
+                        null,
+                        System.currentTimeMillis());
+
+        final ExecutionGraphInfo executionGraphInfo = new ExecutionGraphInfo(executionGraph);
+
+        final ErrorInfo failureInfo =
+                executionGraphInfo.getArchivedExecutionGraph().getFailureInfo();
+
+        final Iterator<RootExceptionHistoryEntry> exceptionHistoryEntryIterator =
+                executionGraphInfo.getExceptionHistory().iterator();
+
+        assertTrue(exceptionHistoryEntryIterator.hasNext());
+        final RootExceptionHistoryEntry actualEntry = exceptionHistoryEntryIterator.next();
+        assertFalse(exceptionHistoryEntryIterator.hasNext());
+
+        assertNotNull(failureInfo);
+        assertEquals(failureInfo.getException(), actualEntry.getException());
+        assertEquals(failureInfo.getTimestamp(), actualEntry.getTimestamp());
+
+        assertTrue(actualEntry.isGlobal());
+        assertFalse(actualEntry.getConcurrentExceptions().iterator().hasNext());
+        assertNull(actualEntry.getFailingTaskName());
+        assertNull(actualEntry.getTaskManagerLocation());
+    }
+}