dmvk commented on a change in pull request #18689: URL: https://github.com/apache/flink/pull/18689#discussion_r803659647
########## File path: flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateWithExecutionGraph.java ########## @@ -148,6 +162,33 @@ public Logger getLogger() { return logger; } + protected Throwable extractError(TaskExecutionStateTransition taskExecutionStateTransition) { + Throwable cause = taskExecutionStateTransition.getError(userCodeClassLoader); + if (cause == null) { + cause = new FlinkException("Unknown failure cause. Probably related to FLINK-21376."); + } + return cause; + } + + protected Optional<ExecutionVertexID> extractExecutionVertexID( + TaskExecutionStateTransition taskExecutionStateTransition) { + return executionGraph.getExecutionVertexId(taskExecutionStateTransition.getID()); Review comment: I'd like to avoid adding new methods to the execution graph. `ExecutionGraph#getRegisteredExecutions` should be enough to cover this use case. Other think would be, that we don't really expect this not to be found ever, so we can throw an exception right away if we don't find an entry. DefaultScheduler does the same thing, even though it's also using optional, it check that the option is not empty after successful update to the execution graph. ########## File path: flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Executing.java ########## @@ -80,49 +89,53 @@ public void cancel() { getExecutionGraph(), getExecutionGraphHandler(), getOperatorCoordinatorHandler()); } + private void handleFailure(Failure failure) { + failureCollection.add(failure); + FailureResult failureResult = context.howToHandleFailure(failure); + transitionOnFailure(failureResult); + } + @Override public void handleGlobalFailure(Throwable cause) { - handleAnyFailure(cause); + handleFailure(Failure.createGlobal(cause)); } - private void handleAnyFailure(Throwable cause) { - final FailureResult failureResult = context.howToHandleFailure(cause); + @Override + boolean updateTaskExecutionState(TaskExecutionStateTransition taskExecutionStateTransition) { + final boolean successfulUpdate = + getExecutionGraph().updateState(taskExecutionStateTransition); + + if (successfulUpdate + && taskExecutionStateTransition.getExecutionState() == ExecutionState.FAILED) { + handleFailure( + Failure.createLocal( + extractError(taskExecutionStateTransition), + extractExecutionVertexID(taskExecutionStateTransition))); + } + + return successfulUpdate; Review comment: This methods is duplicated several times, basically in all states extending `StateWithExecutionGraph` apart from `Cancelling`. 1) It feels that it could be moved up to the base class to avoid code duplication 2) I think tasks can also fail when the job is cancelling (basically when the cancel call on the operator throws an exception). Is this correct? If yes, it would eliminate the need of treating the `Cancelling` state differently. ########## File path: flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateWithExecutionGraph.java ########## @@ -148,6 +162,33 @@ public Logger getLogger() { return logger; } + protected Throwable extractError(TaskExecutionStateTransition taskExecutionStateTransition) { + Throwable cause = taskExecutionStateTransition.getError(userCodeClassLoader); + if (cause == null) { + cause = new FlinkException("Unknown failure cause. Probably related to FLINK-21376."); + } + return cause; + } + + protected Optional<ExecutionVertexID> extractExecutionVertexID( + TaskExecutionStateTransition taskExecutionStateTransition) { + return executionGraph.getExecutionVertexId(taskExecutionStateTransition.getID()); + } + + protected static Optional<RootExceptionHistoryEntry> convertFailures( + Function<ExecutionVertexID, Optional<ExecutionVertex>> lookup, Review comment: This has a weird signature, why can't we simply pass an execution graph here? ########## File path: flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateWithExecutionGraph.java ########## @@ -148,6 +162,33 @@ public Logger getLogger() { return logger; } + protected Throwable extractError(TaskExecutionStateTransition taskExecutionStateTransition) { Review comment: We can get rid of this method once we unify `updateTaskExecutionState` ########## File path: flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/failure/Failure.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.adaptive.failure; + +import org.apache.flink.runtime.executiongraph.ExecutionVertex; +import org.apache.flink.runtime.scheduler.exceptionhistory.ExceptionHistoryEntry; +import org.apache.flink.runtime.scheduler.exceptionhistory.RootExceptionHistoryEntry; +import org.apache.flink.runtime.scheduler.strategy.ExecutionVertexID; + +import java.util.Optional; +import java.util.Set; +import java.util.function.Function; + +/** Failure object. */ +public abstract class Failure { + private final Throwable cause; + private final long timestamp; + + public Failure(Throwable cause) { + this.cause = cause; + this.timestamp = System.currentTimeMillis(); Review comment: Not really, the DefaultlScheduler does the same thing. Unless the timestamp would be reported by the TM, there is not much we can do ########## File path: flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StopWithSavepoint.java ########## @@ -134,30 +143,53 @@ public JobStatus getJobStatus() { return JobStatus.RUNNING; } + private void handleFailure(Failure failure) { + failureCollection.add(failure); + FailureResult failureResult = context.howToHandleFailure(failure); + transitionOnFailure(failureResult); + } + @Override public void handleGlobalFailure(Throwable cause) { - handleAnyFailure(cause); + handleFailure(Failure.createGlobal(cause)); } @Override boolean updateTaskExecutionState(TaskExecutionStateTransition taskExecutionStateTransition) { final boolean successfulUpdate = getExecutionGraph().updateState(taskExecutionStateTransition); - if (successfulUpdate) { - if (taskExecutionStateTransition.getExecutionState() == ExecutionState.FAILED) { - Throwable cause = taskExecutionStateTransition.getError(userCodeClassLoader); - handleAnyFailure( - cause == null - ? new FlinkException( - "Unknown failure cause. Probably related to FLINK-21376.") - : cause); - } + if (successfulUpdate + && taskExecutionStateTransition.getExecutionState() == ExecutionState.FAILED) { + handleFailure( + Failure.createLocal( + extractError(taskExecutionStateTransition), + extractExecutionVertexID(taskExecutionStateTransition))); } return successfulUpdate; } + private void transitionOnFailure(FailureResult failureResult) { Review comment: This is duplicated in `Executing` state. ########## File path: flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateWithExecutionGraph.java ########## @@ -148,6 +162,33 @@ public Logger getLogger() { return logger; } + protected Throwable extractError(TaskExecutionStateTransition taskExecutionStateTransition) { + Throwable cause = taskExecutionStateTransition.getError(userCodeClassLoader); + if (cause == null) { + cause = new FlinkException("Unknown failure cause. Probably related to FLINK-21376."); + } + return cause; + } + + protected Optional<ExecutionVertexID> extractExecutionVertexID( + TaskExecutionStateTransition taskExecutionStateTransition) { + return executionGraph.getExecutionVertexId(taskExecutionStateTransition.getID()); + } + + protected static Optional<RootExceptionHistoryEntry> convertFailures( + Function<ExecutionVertexID, Optional<ExecutionVertex>> lookup, + List<Failure> failureCollection) { + if (failureCollection.isEmpty()) { + return Optional.empty(); + } + Failure first = failureCollection.remove(0); + Set<ExceptionHistoryEntry> entries = new HashSet<>(); + for (Failure failure : failureCollection) { + entries.add(failure.toExceptionHistoryEntry(lookup)); Review comment: Wouldn't simply implementing hashCode & equals for the Failure object do the trick? ########## File path: flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraph.java ########## @@ -1576,4 +1576,15 @@ public ExecutionDeploymentListener getExecutionDeploymentListener() { public boolean isDynamic() { return isDynamic; } + + @Override + public Optional<ExecutionVertexID> getExecutionVertexId(ExecutionAttemptID id) { + Execution execution = this.getRegisteredExecutions().get(id); + return Optional.ofNullable(execution).map(Execution::getVertex).map(ExecutionVertex::getID); + } + + @Override + public Optional<ExecutionVertex> getExecutionVertex(final ExecutionVertexID executionVertexId) { Review comment: We already have `DefaultExecutionGraph#getExecutionVertexOrThrow` in place, so we should avoid adding a new method -- 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. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org