tillrohrmann commented on a change in pull request #14804:
URL: https://github.com/apache/flink/pull/14804#discussion_r579271736
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobManagerRunnerResult.java
##########
@@ -81,21 +83,23 @@ public boolean equals(Object o) {
return false;
}
JobManagerRunnerResult that = (JobManagerRunnerResult) o;
- return Objects.equals(archivedExecutionGraph,
that.archivedExecutionGraph)
+ // TODO: verify ExecutionGraphInfo.equals
Review comment:
still valid?
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/OnCompletionActions.java
##########
@@ -18,18 +18,19 @@
package org.apache.flink.runtime.jobmanager;
-import org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph;
+import org.apache.flink.runtime.executiongraph.AccessExecutionGraph;
import org.apache.flink.runtime.jobmaster.JobMaster;
+import org.apache.flink.runtime.scheduler.ExecutionGraphInfo;
/** Interface for completion actions once a Flink job has reached a terminal
state. */
public interface OnCompletionActions {
/**
* Job reached a globally terminal state.
*
- * @param executionGraph serializable execution graph
+ * @param executionGraphInfo serializable {@link
AccessExecutionGraph}-related information
Review comment:
```suggestion
* @param executionGraphInfo executionGraphInfo contains information
about the terminated job
```
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobManagerRunnerResult.java
##########
@@ -81,21 +83,23 @@ public boolean equals(Object o) {
return false;
}
JobManagerRunnerResult that = (JobManagerRunnerResult) o;
- return Objects.equals(archivedExecutionGraph,
that.archivedExecutionGraph)
+ // TODO: verify ExecutionGraphInfo.equals
+ return Objects.equals(executionGraphInfo, that.executionGraphInfo)
&& Objects.equals(failure, that.failure);
}
@Override
public int hashCode() {
- return Objects.hash(archivedExecutionGraph, failure);
+ // TODO: verify ExecutionGraphInfo.hashCode
Review comment:
still valid?
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobManagerRunnerImpl.java
##########
@@ -241,10 +241,10 @@ public void start() throws Exception {
/** Job completion notification triggered by JobManager. */
@Override
- public void jobReachedGloballyTerminalState(ArchivedExecutionGraph
executionGraph) {
+ public void jobReachedGloballyTerminalState(ExecutionGraphInfo
executionGraphInfo) {
unregisterJobFromHighAvailability();
// complete the result future with the terminal execution graph
Review comment:
Comment seems to be outdated with this change.
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/FileExecutionGraphInfoStoreTest.java
##########
@@ -220,40 +226,46 @@ public void testCloseCleansUp() throws IOException {
assertThat(storageDirectory.listFiles().length,
Matchers.equalTo(0));
executionGraphStore.put(
- new
ArchivedExecutionGraphBuilder().setState(JobStatus.FINISHED).build());
+ new ExecutionGraphInfo(
+ new ArchivedExecutionGraphBuilder()
+ .setState(JobStatus.FINISHED)
+ .build()));
assertThat(storageDirectory.listFiles().length,
Matchers.equalTo(1));
}
assertThat(rootDir.listFiles().length, Matchers.equalTo(0));
}
- /** Tests that evicted {@link ArchivedExecutionGraph} are loaded from disk
again. */
+ /** Tests that evicted {@link ExecutionGraphInfo} are loaded from disk
again. */
@Test
public void testCacheLoading() throws IOException {
final File rootDir = temporaryFolder.newFolder();
- try (final FileArchivedExecutionGraphStore executionGraphStore =
- new FileArchivedExecutionGraphStore(
+ try (final FileExecutionGraphInfoStore executionGraphStore =
+ new FileExecutionGraphInfoStore(
rootDir,
Time.hours(1L),
Integer.MAX_VALUE,
100L << 10,
TestingUtils.defaultScheduledExecutor(),
Ticker.systemTicker())) {
- final LoadingCache<JobID, ArchivedExecutionGraph>
executionGraphCache =
- executionGraphStore.getArchivedExecutionGraphCache();
+ final LoadingCache<JobID, ExecutionGraphInfo> executionGraphCache =
+ executionGraphStore.getExecutionGraphInfoCache();
- Collection<ArchivedExecutionGraph> executionGraphs = new
ArrayList<>(64);
+ Collection<ExecutionGraphInfo> executionGraphs = new
ArrayList<>(64);
Review comment:
```suggestion
Collection<ExecutionGraphInfo> executionGraphInfos = new
ArrayList<>(64);
```
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherTest.java
##########
@@ -530,20 +531,22 @@ public void testCacheJobExecutionResult() throws
Exception {
final JobID failedJobId = new JobID();
final JobStatus expectedState = JobStatus.FAILED;
- final ArchivedExecutionGraph failedExecutionGraph =
- new ArchivedExecutionGraphBuilder()
- .setJobID(failedJobId)
- .setState(expectedState)
- .setFailureCause(new ErrorInfo(new
RuntimeException("expected"), 1L))
- .build();
+ final ExecutionGraphInfo failedExecutionGraph =
Review comment:
```suggestion
final ExecutionGraphInfo failedExecutionGraphInfo =
```
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobManagerRunnerResult.java
##########
@@ -56,9 +57,10 @@ public boolean isInitializationFailure() {
* @return the successful completed {@link ArchivedExecutionGraph}
* @throws IllegalStateException if the result is not a success
*/
- public ArchivedExecutionGraph getArchivedExecutionGraph() {
+ // TODO: rename method and javadoc
Review comment:
I guess you still wanna do this, right?
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/FileExecutionGraphInfoStoreTest.java
##########
@@ -267,18 +279,18 @@ public void testCacheLoading() throws IOException {
assertThat(
storageDirectory.listFiles().length,
Matchers.equalTo(executionGraphs.size()));
- for (ArchivedExecutionGraph executionGraph : executionGraphs) {
+ for (ExecutionGraphInfo executionGraph : executionGraphs) {
Review comment:
```suggestion
for (ExecutionGraphInfo executionGraphInfo :
executionGraphInfos) {
```
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/AbstractArchivedExecutionGraphHandler.java
##########
@@ -0,0 +1,74 @@
+/*
+ * 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.rest.handler.job;
+
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.runtime.executiongraph.AccessExecutionGraph;
+import org.apache.flink.runtime.rest.handler.HandlerRequest;
+import org.apache.flink.runtime.rest.handler.RestHandlerException;
+import org.apache.flink.runtime.rest.handler.legacy.ExecutionGraphCache;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.JobMessageParameters;
+import org.apache.flink.runtime.rest.messages.MessageHeaders;
+import org.apache.flink.runtime.rest.messages.ResponseBody;
+import org.apache.flink.runtime.scheduler.ExecutionGraphInfo;
+import org.apache.flink.runtime.webmonitor.RestfulGateway;
+import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
+
+import java.util.Map;
+import java.util.concurrent.Executor;
+
+/**
+ * {@code AbstractArchivedExecutionGraphHandler} handles requests that require
accessing the job's
+ * {@link AccessExecutionGraph}.
+ *
+ * @param <R> the response type
+ * @param <M> the message parameter type
+ */
+public abstract class AbstractArchivedExecutionGraphHandler<
Review comment:
Maybe call it `AbstractAccessExecutionGraphHandler` since we are giving
access to the `AccessExecutionGraphHandler`.
----------------------------------------------------------------
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]
