XComp commented on a change in pull request #15640: URL: https://github.com/apache/flink/pull/15640#discussion_r615681598
########## File path: flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/exceptionhistory/FailureHandlingResultSnapshot.java ########## @@ -0,0 +1,146 @@ +/* + * 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.exceptionhistory; + +import org.apache.flink.annotation.VisibleForTesting; +import org.apache.flink.runtime.executiongraph.Execution; +import org.apache.flink.runtime.executiongraph.ExecutionAttemptID; +import org.apache.flink.runtime.executiongraph.ExecutionVertex; +import org.apache.flink.runtime.executiongraph.failover.flip1.FailureHandlingResult; +import org.apache.flink.runtime.scheduler.strategy.ExecutionVertexID; +import org.apache.flink.util.Preconditions; + +import javax.annotation.Nullable; + +import java.util.HashMap; +import java.util.Map; +import java.util.Optional; +import java.util.stream.Collectors; + +/** + * {@code FailureHandlingResultSnapshot} creates a snapshot of a {@link FailureHandlingResult} + * providing the actual {@link Execution Executions}. + */ +public class FailureHandlingResultSnapshot { + + private final ExecutionAttemptID rootCauseExecution; + private final Throwable rootCause; + private final long timestamp; + private final Map<ExecutionAttemptID, Execution> failedExecutions; + + /** + * Creates a {@code FailureHandlingResultSnapshot} based on the passed {@link + * FailureHandlingResult} and {@link ExecutionVertex ExecutionVertices}. + * + * @param failureHandlingResult The {@code FailureHandlingResult} that is used for extracting + * the failure information. + * @param executionVertices The {@code ExecutionVertices} that shall be scanned for failures + * based on the passed {@code FailureHandlingResult}. + * @return The {@code FailureHandlingResultSnapshot}. + */ + public static FailureHandlingResultSnapshot create( + FailureHandlingResult failureHandlingResult, + Iterable<ExecutionVertex> executionVertices) { + final ExecutionVertexID rootCauseExecutionVertexId = + failureHandlingResult.getExecutionVertexIdOfFailedTask().orElse(null); + + ExecutionAttemptID rootCauseExecutionAttemptId = null; + final Map<ExecutionAttemptID, Execution> failedExecutions = new HashMap<>(); + for (ExecutionVertex executionVertex : executionVertices) { + if (executionVertex.getID().equals(rootCauseExecutionVertexId)) { + Preconditions.checkArgument( + executionVertex.getCurrentExecutionAttempt().getFailureInfo().isPresent(), + String.format( + "The execution %s didn't provide a failure info even though the corresponding ExecutionVertex %s is marked as having handled the root cause of this failure.", + executionVertex.getCurrentExecutionAttempt().getAttemptId(), + executionVertex.getID())); + rootCauseExecutionAttemptId = + executionVertex.getCurrentExecutionAttempt().getAttemptId(); + } + + final Execution execution = executionVertex.getCurrentExecutionAttempt(); + if (failureHandlingResult.getVerticesToRestart().contains(executionVertex.getID()) + && execution.getFailureInfo().isPresent()) { + failedExecutions.put(execution.getAttemptId(), execution); + } + } + + return new FailureHandlingResultSnapshot( + rootCauseExecutionAttemptId, + failureHandlingResult.getError(), Review comment: > I added a workaround setting the exception explicitly to have a working version. The issue in general is again what's already covered by FLINK-22060. Without the workaround, JobMasterTest is failing. AFAIR, it is a known issue that came up while doing the research on FLINK-22060. The JobMasterTest does not set the failure cause (because it didn't need to before). > The workaround fixes it for now and we should address the issue in a follow up ticket before releasing. I re-iterated over the `@Nullable` constraints and updated the code accordingly. We're relying on a utility method `ErrorInfo.handleMissingThrowable` that addresses the issue of [FLINK-21376](https://issues.apache.org/jira/browse/FLINK-21376)/[FLINK-22060](https://issues.apache.org/jira/browse/FLINK-22060) for now. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: [email protected]
