You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2021/02/17 09:31:24 UTC
[GitHub] [flink] rmetzger opened a new pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
rmetzger opened a new pull request #14950:
URL: https://github.com/apache/flink/pull/14950
## What is the purpose of the change
Mocking the `ExecutionGraph` is very difficult because of all its dependencies. This PR proposes to extract an interface from the ExecutionGraph.
## Brief change log
- Extract interface
- adjust *State test mocks
- introduce mock base
## Verifying this change
Covered by existing tests.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #14950:
URL: https://github.com/apache/flink/pull/14950#issuecomment-780431068
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* e44c9cdb2e60872179635616e8ffb184bb0337b9 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14950:
URL: https://github.com/apache/flink/pull/14950#issuecomment-780431068
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13398",
"triggerID" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"triggerType" : "PUSH"
}, {
"hash" : "bee61356ebeb5f99c96ce3193b3bd8205e61e44d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13409",
"triggerID" : "bee61356ebeb5f99c96ce3193b3bd8205e61e44d",
"triggerType" : "PUSH"
}, {
"hash" : "1b43a411d22af19f12f14a79c1094c8dc5cfeb42",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "1b43a411d22af19f12f14a79c1094c8dc5cfeb42",
"triggerType" : "PUSH"
}, {
"hash" : "e6aae42cf093d4f44b93b79a78bdc64538df8725",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13445",
"triggerID" : "e6aae42cf093d4f44b93b79a78bdc64538df8725",
"triggerType" : "PUSH"
}, {
"hash" : "cccca8191e6800961467c726695c8ef7575d157a",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13449",
"triggerID" : "cccca8191e6800961467c726695c8ef7575d157a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1b43a411d22af19f12f14a79c1094c8dc5cfeb42 UNKNOWN
* cccca8191e6800961467c726695c8ef7575d157a Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13449)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14950:
URL: https://github.com/apache/flink/pull/14950#issuecomment-780431068
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13398",
"triggerID" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"triggerType" : "PUSH"
}, {
"hash" : "bee61356ebeb5f99c96ce3193b3bd8205e61e44d",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13409",
"triggerID" : "bee61356ebeb5f99c96ce3193b3bd8205e61e44d",
"triggerType" : "PUSH"
}, {
"hash" : "1b43a411d22af19f12f14a79c1094c8dc5cfeb42",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "1b43a411d22af19f12f14a79c1094c8dc5cfeb42",
"triggerType" : "PUSH"
}, {
"hash" : "e6aae42cf093d4f44b93b79a78bdc64538df8725",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13445",
"triggerID" : "e6aae42cf093d4f44b93b79a78bdc64538df8725",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* bee61356ebeb5f99c96ce3193b3bd8205e61e44d Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13409)
* 1b43a411d22af19f12f14a79c1094c8dc5cfeb42 UNKNOWN
* e6aae42cf093d4f44b93b79a78bdc64538df8725 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13445)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] zentol commented on a change in pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
zentol commented on a change in pull request #14950:
URL: https://github.com/apache/flink/pull/14950#discussion_r578975860
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/declarative/StateTrackingMockExecutionGraph.java
##########
@@ -0,0 +1,459 @@
+/*
+ * 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.declarative;
+
+import org.apache.flink.api.common.ArchivedExecutionConfig;
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.api.common.accumulators.Accumulator;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.JobException;
+import org.apache.flink.runtime.accumulators.AccumulatorSnapshot;
+import org.apache.flink.runtime.accumulators.StringifiedAccumulatorResult;
+import org.apache.flink.runtime.blob.BlobWriter;
+import org.apache.flink.runtime.blob.PermanentBlobKey;
+import org.apache.flink.runtime.checkpoint.CheckpointCoordinator;
+import org.apache.flink.runtime.checkpoint.CheckpointIDCounter;
+import org.apache.flink.runtime.checkpoint.CheckpointStatsSnapshot;
+import org.apache.flink.runtime.checkpoint.CheckpointStatsTracker;
+import org.apache.flink.runtime.checkpoint.CheckpointsCleaner;
+import org.apache.flink.runtime.checkpoint.CompletedCheckpointStore;
+import org.apache.flink.runtime.checkpoint.MasterTriggerRestoreHook;
+import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor;
+import org.apache.flink.runtime.execution.ExecutionState;
+import org.apache.flink.runtime.executiongraph.ErrorInfo;
+import org.apache.flink.runtime.executiongraph.Execution;
+import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
+import org.apache.flink.runtime.executiongraph.ExecutionDeploymentListener;
+import org.apache.flink.runtime.executiongraph.ExecutionGraph;
+import org.apache.flink.runtime.executiongraph.ExecutionJobVertex;
+import org.apache.flink.runtime.executiongraph.ExecutionVertex;
+import org.apache.flink.runtime.executiongraph.IntermediateResult;
+import org.apache.flink.runtime.executiongraph.JobInformation;
+import org.apache.flink.runtime.executiongraph.JobStatusListener;
+import org.apache.flink.runtime.executiongraph.TaskExecutionStateTransition;
+import org.apache.flink.runtime.executiongraph.failover.flip1.ResultPartitionAvailabilityChecker;
+import org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease.PartitionReleaseStrategy;
+import org.apache.flink.runtime.io.network.partition.JobMasterPartitionTracker;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
+import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
+import org.apache.flink.runtime.jobgraph.JobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+import org.apache.flink.runtime.jobgraph.ScheduleMode;
+import org.apache.flink.runtime.jobgraph.tasks.CheckpointCoordinatorConfiguration;
+import org.apache.flink.runtime.query.KvStateLocationRegistry;
+import org.apache.flink.runtime.scheduler.InternalFailuresListener;
+import org.apache.flink.runtime.scheduler.strategy.SchedulingTopology;
+import org.apache.flink.runtime.shuffle.ShuffleMaster;
+import org.apache.flink.runtime.state.CheckpointStorage;
+import org.apache.flink.runtime.state.StateBackend;
+import org.apache.flink.types.Either;
+import org.apache.flink.util.OptionalFailure;
+import org.apache.flink.util.SerializedValue;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ForkJoinPool;
+
+/**
+ * Mocked ExecutionGraph which (partially) tracks the job status, and provides some basic mocks to
+ * create an {@link org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph} from this
+ * ExecutionGraph.
+ */
+class StateTrackingMockExecutionGraph implements ExecutionGraph {
+ private JobStatus state = JobStatus.INITIALIZING;
+ private final CompletableFuture<JobStatus> terminationFuture = new CompletableFuture<>();
+ private final JobID jobId = new JobID();
+ private final ArchivedExecutionConfig archivedExecutionConfig = new ExecutionConfig().archive();
+
+ // ---- methods to control the mock
+
+ void completeTerminationFuture(JobStatus finalStatus) {
+ terminationFuture.complete(finalStatus);
+ state = finalStatus;
+ }
+
+ // ---- interface implementations
+
+ @Override
+ public boolean updateState(TaskExecutionStateTransition state) {
+ return true;
+ }
+
+ @Override
+ public JobStatus getState() {
+ return state;
+ }
+
+ @Override
+ public CompletableFuture<JobStatus> getTerminationFuture() {
+ return terminationFuture;
+ }
+
+ @Override
+ public Executor getFutureExecutor() {
+ return ForkJoinPool.commonPool();
+ }
+
+ @Override
+ public void registerExecution(Execution exec) {
+ // no-op
+ }
+
+ @Override
+ public void cancel() {
+ state = JobStatus.CANCELLING;
+ }
+
+ @Override
+ public void failJob(Throwable cause) {
+ state = JobStatus.FAILING;
+ }
+
+ @Override
+ public void suspend(Throwable suspensionCause) {
+ state = JobStatus.SUSPENDED;
+ }
+
+ @Override
+ public void transitionToRunning() {
+ state = JobStatus.RUNNING;
+ }
+
+ // --- interface implementations: methods for creating an archived execution graph
+
+ @Override
+ public int getTotalNumberOfVertices() {
+ return 0;
+ }
+
+ @Override
+ public Iterable<ExecutionJobVertex> getVerticesTopologically() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public Map<JobVertexID, ExecutionJobVertex> getAllVertices() {
+ return Collections.emptyMap();
+ }
+
+ @Override
+ public Map<String, SerializedValue<OptionalFailure<Object>>> getAccumulatorsSerialized() {
+ return Collections.emptyMap();
+ }
+
+ @Override
+ public long getStatusTimestamp(JobStatus status) {
+ return 0L;
+ }
+
+ @Override
+ public String getJsonPlan() {
+ return "";
+ }
+
+ @Override
+ public JobID getJobID() {
+ return jobId;
+ }
+
+ @Override
+ public String getJobName() {
+ return "testJob";
+ }
+
+ @Nullable
+ @Override
+ public ErrorInfo getFailureInfo() {
+ return null;
+ }
+
+ @Nullable
+ @Override
+ public ArchivedExecutionConfig getArchivedExecutionConfig() {
+ return archivedExecutionConfig;
+ }
+
+ @Override
+ public boolean isStoppable() {
+ return false;
+ }
+
+ @Nullable
+ @Override
+ public CheckpointCoordinatorConfiguration getCheckpointCoordinatorConfiguration() {
+ return null;
+ }
+
+ @Nullable
+ @Override
+ public CheckpointStatsSnapshot getCheckpointStatsSnapshot() {
+ return null;
+ }
+
+ @Override
+ public Optional<String> getStateBackendName() {
+ return Optional.empty();
+ }
+
+ @Override
+ public Optional<String> getCheckpointStorageName() {
+ return Optional.empty();
+ }
+
+ @Override
+ public StringifiedAccumulatorResult[] getAccumulatorResultsStringified() {
+ return new StringifiedAccumulatorResult[0];
+ }
+
+ // -- remaining interface implementations: all unsupported
+
+ @Override
+ public boolean isArchived() {
+ throw new UnsupportedOperationException();
Review comment:
we could remove this method entirely
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] rmetzger commented on a change in pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
rmetzger commented on a change in pull request #14950:
URL: https://github.com/apache/flink/pull/14950#discussion_r580511110
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
##########
@@ -1021,203 +220,20 @@ public ArchivedExecutionConfig getArchivedExecutionConfig() {
*
* @return Termination future of this {@link ExecutionGraph}.
*/
- public CompletableFuture<JobStatus> getTerminationFuture() {
- return terminationFuture;
- }
+ CompletableFuture<JobStatus> getTerminationFuture();
@VisibleForTesting
- public JobStatus waitUntilTerminal() throws InterruptedException {
- try {
- return terminationFuture.get();
- } catch (ExecutionException e) {
- // this should never happen
- // it would be a bug, so we don't expect this to be handled and throw
- // an unchecked exception here
- throw new RuntimeException(e);
- }
- }
-
- // ------------------------------------------------------------------------
- // State Transitions
- // ------------------------------------------------------------------------
-
- public boolean transitionState(JobStatus current, JobStatus newState) {
- return transitionState(current, newState, null);
- }
-
- private void transitionState(JobStatus newState, Throwable error) {
- transitionState(state, newState, error);
- }
+ JobStatus waitUntilTerminal() throws InterruptedException;
- private boolean transitionState(JobStatus current, JobStatus newState, Throwable error) {
- assertRunningInJobMasterMainThread();
- // consistency check
- if (current.isTerminalState()) {
- String message = "Job is trying to leave terminal state " + current;
- LOG.error(message);
- throw new IllegalStateException(message);
- }
+ boolean transitionState(JobStatus current, JobStatus newState);
- // now do the actual state transition
- if (state == current) {
- state = newState;
- LOG.info(
- "Job {} ({}) switched from state {} to {}.",
- getJobName(),
- getJobID(),
- current,
- newState,
- error);
+ void incrementRestarts();
- stateTimestamps[newState.ordinal()] = System.currentTimeMillis();
- notifyJobStatusChange(newState, error);
- return true;
- } else {
- return false;
- }
- }
+ void initFailureCause(Throwable t);
- public void incrementRestarts() {
- numberOfRestartsCounter.inc();
- }
+ void vertexFinished();
Review comment:
Sorry to ask again, but I don't seem to understand your suggestion here (and I'm having the feeling I'm overlooking something obvious)
None of the package-private methods are used in the context of the ExecutionGraph interface. The only method I'm using in the `Executing.deploy()` method is `getVerticesTopologically()`, which was public before.
deploy/markFailed are private methods of `ExecutionVertex`, which I don't expose in this PR.
Introducing an interface for `ExecutionJobVertex` (with a `.getDeployableTaskVertices()`) and `ExecutionVertex` (with a `.deploy()` and `.markFailed()` ) would indeed simplify this test, but seems out of scope of this change.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] rmetzger commented on a change in pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
rmetzger commented on a change in pull request #14950:
URL: https://github.com/apache/flink/pull/14950#discussion_r577640124
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/MockExecutionGraphBase.java
##########
@@ -0,0 +1,424 @@
+/*
+ * 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.executiongraph;
+
+import org.apache.flink.api.common.ArchivedExecutionConfig;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.api.common.accumulators.Accumulator;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.JobException;
+import org.apache.flink.runtime.accumulators.AccumulatorSnapshot;
+import org.apache.flink.runtime.accumulators.StringifiedAccumulatorResult;
+import org.apache.flink.runtime.blob.BlobWriter;
+import org.apache.flink.runtime.blob.PermanentBlobKey;
+import org.apache.flink.runtime.checkpoint.CheckpointCoordinator;
+import org.apache.flink.runtime.checkpoint.CheckpointIDCounter;
+import org.apache.flink.runtime.checkpoint.CheckpointStatsSnapshot;
+import org.apache.flink.runtime.checkpoint.CheckpointStatsTracker;
+import org.apache.flink.runtime.checkpoint.CheckpointsCleaner;
+import org.apache.flink.runtime.checkpoint.CompletedCheckpointStore;
+import org.apache.flink.runtime.checkpoint.MasterTriggerRestoreHook;
+import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor;
+import org.apache.flink.runtime.execution.ExecutionState;
+import org.apache.flink.runtime.executiongraph.failover.flip1.ResultPartitionAvailabilityChecker;
+import org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease.PartitionReleaseStrategy;
+import org.apache.flink.runtime.io.network.partition.JobMasterPartitionTracker;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
+import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
+import org.apache.flink.runtime.jobgraph.JobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+import org.apache.flink.runtime.jobgraph.ScheduleMode;
+import org.apache.flink.runtime.jobgraph.tasks.CheckpointCoordinatorConfiguration;
+import org.apache.flink.runtime.query.KvStateLocationRegistry;
+import org.apache.flink.runtime.scheduler.InternalFailuresListener;
+import org.apache.flink.runtime.scheduler.strategy.SchedulingTopology;
+import org.apache.flink.runtime.shuffle.ShuffleMaster;
+import org.apache.flink.runtime.state.CheckpointStorage;
+import org.apache.flink.runtime.state.StateBackend;
+import org.apache.flink.types.Either;
+import org.apache.flink.util.OptionalFailure;
+import org.apache.flink.util.SerializedValue;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+
+/** Base class for mocking the ExecutionGraph. */
+public class MockExecutionGraphBase implements ExecutionGraph {
+
+ @Override
+ public String getJsonPlan() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public JobID getJobID() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public String getJobName() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public JobStatus getState() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Nullable
+ @Override
+ public ErrorInfo getFailureInfo() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public long getStatusTimestamp(JobStatus status) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Nullable
+ @Override
+ public CheckpointCoordinatorConfiguration getCheckpointCoordinatorConfiguration() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Nullable
+ @Override
+ public CheckpointStatsSnapshot getCheckpointStatsSnapshot() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Nullable
+ @Override
+ public ArchivedExecutionConfig getArchivedExecutionConfig() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean isStoppable() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public StringifiedAccumulatorResult[] getAccumulatorResultsStringified() {
+ return new StringifiedAccumulatorResult[0];
+ }
+
+ @Override
+ public Map<String, SerializedValue<OptionalFailure<Object>>> getAccumulatorsSerialized() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean isArchived() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Optional<String> getStateBackendName() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Optional<String> getCheckpointStorageName() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void start(@Nonnull ComponentMainThreadExecutor jobMasterMainThreadExecutor) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public int getNumberOfExecutionJobVertices() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public SchedulingTopology getSchedulingTopology() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public ScheduleMode getScheduleMode() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Nonnull
+ @Override
+ public ComponentMainThreadExecutor getJobMasterMainThreadExecutor() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void enableCheckpointing(
+ CheckpointCoordinatorConfiguration chkConfig,
+ List<MasterTriggerRestoreHook<?>> masterHooks,
+ CheckpointIDCounter checkpointIDCounter,
+ CompletedCheckpointStore checkpointStore,
+ StateBackend checkpointStateBackend,
+ CheckpointStorage checkpointStorage,
+ CheckpointStatsTracker statsTracker,
+ CheckpointsCleaner checkpointsCleaner) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Nullable
+ @Override
+ public CheckpointCoordinator getCheckpointCoordinator() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public KvStateLocationRegistry getKvStateLocationRegistry() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void setJsonPlan(String jsonPlan) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Either<SerializedValue<JobInformation>, PermanentBlobKey> getJobInformationOrBlobKey() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Configuration getJobConfiguration() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public ClassLoader getUserClassLoader() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Throwable getFailureCause() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Iterable<ExecutionJobVertex> getVerticesTopologically() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Iterable<ExecutionVertex> getAllExecutionVertices() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public ExecutionJobVertex getJobVertex(JobVertexID id) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Map<JobVertexID, ExecutionJobVertex> getAllVertices() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public long getNumberOfRestarts() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public int getTotalNumberOfVertices() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Map<IntermediateDataSetID, IntermediateResult> getAllIntermediateResults() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public BlobWriter getBlobWriter() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Executor getFutureExecutor() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Map<String, OptionalFailure<Accumulator<?, ?>>> aggregateUserAccumulators() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void setInternalTaskFailuresListener(
+ InternalFailuresListener internalTaskFailuresListener) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void attachJobGraph(List<JobVertex> topologiallySorted) throws JobException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void transitionToRunning() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void cancel() {
+ throw new UnsupportedOperationException();
Review comment:
For my use-case, I'd almost say the opposite is true. I consider it useful to see right away what I've missed to implement in my mock by having an exception thrown. Making cancel a no-op would require me to first understand why the cancellation hasn't happened as expected. With an exception, I notice it right away.
The MockExecutionGraphBase is really just a utility for the state tests. I agree that it might be of debatable use for others. Would moving the MockBase to the declarative scheduler tests package resolve your concern?
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] zentol commented on a change in pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
zentol commented on a change in pull request #14950:
URL: https://github.com/apache/flink/pull/14950#discussion_r578976098
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/declarative/StateTrackingMockExecutionGraph.java
##########
@@ -0,0 +1,459 @@
+/*
+ * 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.declarative;
+
+import org.apache.flink.api.common.ArchivedExecutionConfig;
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.api.common.accumulators.Accumulator;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.JobException;
+import org.apache.flink.runtime.accumulators.AccumulatorSnapshot;
+import org.apache.flink.runtime.accumulators.StringifiedAccumulatorResult;
+import org.apache.flink.runtime.blob.BlobWriter;
+import org.apache.flink.runtime.blob.PermanentBlobKey;
+import org.apache.flink.runtime.checkpoint.CheckpointCoordinator;
+import org.apache.flink.runtime.checkpoint.CheckpointIDCounter;
+import org.apache.flink.runtime.checkpoint.CheckpointStatsSnapshot;
+import org.apache.flink.runtime.checkpoint.CheckpointStatsTracker;
+import org.apache.flink.runtime.checkpoint.CheckpointsCleaner;
+import org.apache.flink.runtime.checkpoint.CompletedCheckpointStore;
+import org.apache.flink.runtime.checkpoint.MasterTriggerRestoreHook;
+import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor;
+import org.apache.flink.runtime.execution.ExecutionState;
+import org.apache.flink.runtime.executiongraph.ErrorInfo;
+import org.apache.flink.runtime.executiongraph.Execution;
+import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
+import org.apache.flink.runtime.executiongraph.ExecutionDeploymentListener;
+import org.apache.flink.runtime.executiongraph.ExecutionGraph;
+import org.apache.flink.runtime.executiongraph.ExecutionJobVertex;
+import org.apache.flink.runtime.executiongraph.ExecutionVertex;
+import org.apache.flink.runtime.executiongraph.IntermediateResult;
+import org.apache.flink.runtime.executiongraph.JobInformation;
+import org.apache.flink.runtime.executiongraph.JobStatusListener;
+import org.apache.flink.runtime.executiongraph.TaskExecutionStateTransition;
+import org.apache.flink.runtime.executiongraph.failover.flip1.ResultPartitionAvailabilityChecker;
+import org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease.PartitionReleaseStrategy;
+import org.apache.flink.runtime.io.network.partition.JobMasterPartitionTracker;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
+import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
+import org.apache.flink.runtime.jobgraph.JobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+import org.apache.flink.runtime.jobgraph.ScheduleMode;
+import org.apache.flink.runtime.jobgraph.tasks.CheckpointCoordinatorConfiguration;
+import org.apache.flink.runtime.query.KvStateLocationRegistry;
+import org.apache.flink.runtime.scheduler.InternalFailuresListener;
+import org.apache.flink.runtime.scheduler.strategy.SchedulingTopology;
+import org.apache.flink.runtime.shuffle.ShuffleMaster;
+import org.apache.flink.runtime.state.CheckpointStorage;
+import org.apache.flink.runtime.state.StateBackend;
+import org.apache.flink.types.Either;
+import org.apache.flink.util.OptionalFailure;
+import org.apache.flink.util.SerializedValue;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ForkJoinPool;
+
+/**
+ * Mocked ExecutionGraph which (partially) tracks the job status, and provides some basic mocks to
+ * create an {@link org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph} from this
+ * ExecutionGraph.
+ */
+class StateTrackingMockExecutionGraph implements ExecutionGraph {
+ private JobStatus state = JobStatus.INITIALIZING;
+ private final CompletableFuture<JobStatus> terminationFuture = new CompletableFuture<>();
+ private final JobID jobId = new JobID();
+ private final ArchivedExecutionConfig archivedExecutionConfig = new ExecutionConfig().archive();
+
+ // ---- methods to control the mock
+
+ void completeTerminationFuture(JobStatus finalStatus) {
+ terminationFuture.complete(finalStatus);
+ state = finalStatus;
+ }
+
+ // ---- interface implementations
+
+ @Override
+ public boolean updateState(TaskExecutionStateTransition state) {
+ return true;
+ }
+
+ @Override
+ public JobStatus getState() {
+ return state;
+ }
+
+ @Override
+ public CompletableFuture<JobStatus> getTerminationFuture() {
+ return terminationFuture;
+ }
+
+ @Override
+ public Executor getFutureExecutor() {
+ return ForkJoinPool.commonPool();
+ }
+
+ @Override
+ public void registerExecution(Execution exec) {
+ // no-op
+ }
+
+ @Override
+ public void cancel() {
+ state = JobStatus.CANCELLING;
+ }
+
+ @Override
+ public void failJob(Throwable cause) {
+ state = JobStatus.FAILING;
+ }
+
+ @Override
+ public void suspend(Throwable suspensionCause) {
+ state = JobStatus.SUSPENDED;
+ }
+
+ @Override
+ public void transitionToRunning() {
+ state = JobStatus.RUNNING;
+ }
+
+ // --- interface implementations: methods for creating an archived execution graph
+
+ @Override
+ public int getTotalNumberOfVertices() {
+ return 0;
+ }
+
+ @Override
+ public Iterable<ExecutionJobVertex> getVerticesTopologically() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public Map<JobVertexID, ExecutionJobVertex> getAllVertices() {
+ return Collections.emptyMap();
+ }
+
+ @Override
+ public Map<String, SerializedValue<OptionalFailure<Object>>> getAccumulatorsSerialized() {
+ return Collections.emptyMap();
+ }
+
+ @Override
+ public long getStatusTimestamp(JobStatus status) {
+ return 0L;
+ }
+
+ @Override
+ public String getJsonPlan() {
+ return "";
+ }
+
+ @Override
+ public JobID getJobID() {
+ return jobId;
+ }
+
+ @Override
+ public String getJobName() {
+ return "testJob";
+ }
+
+ @Nullable
+ @Override
+ public ErrorInfo getFailureInfo() {
+ return null;
+ }
+
+ @Nullable
+ @Override
+ public ArchivedExecutionConfig getArchivedExecutionConfig() {
+ return archivedExecutionConfig;
+ }
+
+ @Override
+ public boolean isStoppable() {
+ return false;
+ }
+
+ @Nullable
+ @Override
+ public CheckpointCoordinatorConfiguration getCheckpointCoordinatorConfiguration() {
+ return null;
+ }
+
+ @Nullable
+ @Override
+ public CheckpointStatsSnapshot getCheckpointStatsSnapshot() {
+ return null;
+ }
+
+ @Override
+ public Optional<String> getStateBackendName() {
+ return Optional.empty();
+ }
+
+ @Override
+ public Optional<String> getCheckpointStorageName() {
+ return Optional.empty();
+ }
+
+ @Override
+ public StringifiedAccumulatorResult[] getAccumulatorResultsStringified() {
+ return new StringifiedAccumulatorResult[0];
+ }
+
+ // -- remaining interface implementations: all unsupported
+
+ @Override
+ public boolean isArchived() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void start(@Nonnull ComponentMainThreadExecutor jobMasterMainThreadExecutor) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public int getNumberOfExecutionJobVertices() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public SchedulingTopology getSchedulingTopology() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public ScheduleMode getScheduleMode() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Nonnull
+ @Override
+ public ComponentMainThreadExecutor getJobMasterMainThreadExecutor() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void enableCheckpointing(
+ CheckpointCoordinatorConfiguration chkConfig,
+ List<MasterTriggerRestoreHook<?>> masterHooks,
+ CheckpointIDCounter checkpointIDCounter,
+ CompletedCheckpointStore checkpointStore,
+ StateBackend checkpointStateBackend,
+ CheckpointStorage checkpointStorage,
+ CheckpointStatsTracker statsTracker,
+ CheckpointsCleaner checkpointsCleaner) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Nullable
+ @Override
+ public CheckpointCoordinator getCheckpointCoordinator() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public KvStateLocationRegistry getKvStateLocationRegistry() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void setJsonPlan(String jsonPlan) {
+ throw new UnsupportedOperationException();
Review comment:
We could implement this in a simple way
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] zentol commented on a change in pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
zentol commented on a change in pull request #14950:
URL: https://github.com/apache/flink/pull/14950#discussion_r578794106
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/MockExecutionGraphBase.java
##########
@@ -0,0 +1,424 @@
+/*
+ * 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.executiongraph;
+
+import org.apache.flink.api.common.ArchivedExecutionConfig;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.api.common.accumulators.Accumulator;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.JobException;
+import org.apache.flink.runtime.accumulators.AccumulatorSnapshot;
+import org.apache.flink.runtime.accumulators.StringifiedAccumulatorResult;
+import org.apache.flink.runtime.blob.BlobWriter;
+import org.apache.flink.runtime.blob.PermanentBlobKey;
+import org.apache.flink.runtime.checkpoint.CheckpointCoordinator;
+import org.apache.flink.runtime.checkpoint.CheckpointIDCounter;
+import org.apache.flink.runtime.checkpoint.CheckpointStatsSnapshot;
+import org.apache.flink.runtime.checkpoint.CheckpointStatsTracker;
+import org.apache.flink.runtime.checkpoint.CheckpointsCleaner;
+import org.apache.flink.runtime.checkpoint.CompletedCheckpointStore;
+import org.apache.flink.runtime.checkpoint.MasterTriggerRestoreHook;
+import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor;
+import org.apache.flink.runtime.execution.ExecutionState;
+import org.apache.flink.runtime.executiongraph.failover.flip1.ResultPartitionAvailabilityChecker;
+import org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease.PartitionReleaseStrategy;
+import org.apache.flink.runtime.io.network.partition.JobMasterPartitionTracker;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
+import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
+import org.apache.flink.runtime.jobgraph.JobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+import org.apache.flink.runtime.jobgraph.ScheduleMode;
+import org.apache.flink.runtime.jobgraph.tasks.CheckpointCoordinatorConfiguration;
+import org.apache.flink.runtime.query.KvStateLocationRegistry;
+import org.apache.flink.runtime.scheduler.InternalFailuresListener;
+import org.apache.flink.runtime.scheduler.strategy.SchedulingTopology;
+import org.apache.flink.runtime.shuffle.ShuffleMaster;
+import org.apache.flink.runtime.state.CheckpointStorage;
+import org.apache.flink.runtime.state.StateBackend;
+import org.apache.flink.types.Either;
+import org.apache.flink.util.OptionalFailure;
+import org.apache.flink.util.SerializedValue;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+
+/** Base class for mocking the ExecutionGraph. */
+public class MockExecutionGraphBase implements ExecutionGraph {
+
+ @Override
+ public String getJsonPlan() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public JobID getJobID() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public String getJobName() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public JobStatus getState() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Nullable
+ @Override
+ public ErrorInfo getFailureInfo() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public long getStatusTimestamp(JobStatus status) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Nullable
+ @Override
+ public CheckpointCoordinatorConfiguration getCheckpointCoordinatorConfiguration() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Nullable
+ @Override
+ public CheckpointStatsSnapshot getCheckpointStatsSnapshot() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Nullable
+ @Override
+ public ArchivedExecutionConfig getArchivedExecutionConfig() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean isStoppable() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public StringifiedAccumulatorResult[] getAccumulatorResultsStringified() {
+ return new StringifiedAccumulatorResult[0];
+ }
+
+ @Override
+ public Map<String, SerializedValue<OptionalFailure<Object>>> getAccumulatorsSerialized() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean isArchived() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Optional<String> getStateBackendName() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Optional<String> getCheckpointStorageName() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void start(@Nonnull ComponentMainThreadExecutor jobMasterMainThreadExecutor) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public int getNumberOfExecutionJobVertices() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public SchedulingTopology getSchedulingTopology() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public ScheduleMode getScheduleMode() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Nonnull
+ @Override
+ public ComponentMainThreadExecutor getJobMasterMainThreadExecutor() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void enableCheckpointing(
+ CheckpointCoordinatorConfiguration chkConfig,
+ List<MasterTriggerRestoreHook<?>> masterHooks,
+ CheckpointIDCounter checkpointIDCounter,
+ CompletedCheckpointStore checkpointStore,
+ StateBackend checkpointStateBackend,
+ CheckpointStorage checkpointStorage,
+ CheckpointStatsTracker statsTracker,
+ CheckpointsCleaner checkpointsCleaner) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Nullable
+ @Override
+ public CheckpointCoordinator getCheckpointCoordinator() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public KvStateLocationRegistry getKvStateLocationRegistry() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void setJsonPlan(String jsonPlan) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Either<SerializedValue<JobInformation>, PermanentBlobKey> getJobInformationOrBlobKey() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Configuration getJobConfiguration() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public ClassLoader getUserClassLoader() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Throwable getFailureCause() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Iterable<ExecutionJobVertex> getVerticesTopologically() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Iterable<ExecutionVertex> getAllExecutionVertices() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public ExecutionJobVertex getJobVertex(JobVertexID id) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Map<JobVertexID, ExecutionJobVertex> getAllVertices() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public long getNumberOfRestarts() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public int getTotalNumberOfVertices() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Map<IntermediateDataSetID, IntermediateResult> getAllIntermediateResults() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public BlobWriter getBlobWriter() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Executor getFutureExecutor() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Map<String, OptionalFailure<Accumulator<?, ?>>> aggregateUserAccumulators() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void setInternalTaskFailuresListener(
+ InternalFailuresListener internalTaskFailuresListener) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void attachJobGraph(List<JobVertex> topologiallySorted) throws JobException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void transitionToRunning() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void cancel() {
+ throw new UnsupportedOperationException();
Review comment:
For reference, [this](https://github.com/apache/flink/pull/14963/files#r578791931) is exactly the kind of inconvenience that I' concerned about.
I was about the suggest a compromise where certain usually non-interesting methods would be mocked in a non-failing way, but you already had that idea earlier it appears 👍
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] zentol commented on pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
zentol commented on pull request #14950:
URL: https://github.com/apache/flink/pull/14950#issuecomment-788906901
```
Files with unapproved licenses:
flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/MockExecutionGraph.java
```
`git rm flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/MockExecutionGraph.java`
(This is some really funky stuff because I don't even see it in my filesystem)
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] zentol commented on a change in pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
zentol commented on a change in pull request #14950:
URL: https://github.com/apache/flink/pull/14950#discussion_r581861367
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
##########
@@ -1021,203 +220,20 @@ public ArchivedExecutionConfig getArchivedExecutionConfig() {
*
* @return Termination future of this {@link ExecutionGraph}.
*/
- public CompletableFuture<JobStatus> getTerminationFuture() {
- return terminationFuture;
- }
+ CompletableFuture<JobStatus> getTerminationFuture();
@VisibleForTesting
- public JobStatus waitUntilTerminal() throws InterruptedException {
- try {
- return terminationFuture.get();
- } catch (ExecutionException e) {
- // this should never happen
- // it would be a bug, so we don't expect this to be handled and throw
- // an unchecked exception here
- throw new RuntimeException(e);
- }
- }
-
- // ------------------------------------------------------------------------
- // State Transitions
- // ------------------------------------------------------------------------
-
- public boolean transitionState(JobStatus current, JobStatus newState) {
- return transitionState(current, newState, null);
- }
-
- private void transitionState(JobStatus newState, Throwable error) {
- transitionState(state, newState, error);
- }
+ JobStatus waitUntilTerminal() throws InterruptedException;
- private boolean transitionState(JobStatus current, JobStatus newState, Throwable error) {
- assertRunningInJobMasterMainThread();
- // consistency check
- if (current.isTerminalState()) {
- String message = "Job is trying to leave terminal state " + current;
- LOG.error(message);
- throw new IllegalStateException(message);
- }
+ boolean transitionState(JobStatus current, JobStatus newState);
- // now do the actual state transition
- if (state == current) {
- state = newState;
- LOG.info(
- "Job {} ({}) switched from state {} to {}.",
- getJobName(),
- getJobID(),
- current,
- newState,
- error);
+ void incrementRestarts();
- stateTimestamps[newState.ordinal()] = System.currentTimeMillis();
- notifyJobStatusChange(newState, error);
- return true;
- } else {
- return false;
- }
- }
+ void initFailureCause(Throwable t);
- public void incrementRestarts() {
- numberOfRestartsCounter.inc();
- }
+ void vertexFinished();
Review comment:
To add on to that, in the long-term interfaces for all Execution* classes would be pretty sick because we could re-implement the EG as we see fit, whereas an interface between the EG and vertices is probably not that beneficial.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14950:
URL: https://github.com/apache/flink/pull/14950#issuecomment-780431068
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13398",
"triggerID" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"triggerType" : "PUSH"
}, {
"hash" : "bee61356ebeb5f99c96ce3193b3bd8205e61e44d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13409",
"triggerID" : "bee61356ebeb5f99c96ce3193b3bd8205e61e44d",
"triggerType" : "PUSH"
}, {
"hash" : "1b43a411d22af19f12f14a79c1094c8dc5cfeb42",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "1b43a411d22af19f12f14a79c1094c8dc5cfeb42",
"triggerType" : "PUSH"
}, {
"hash" : "e6aae42cf093d4f44b93b79a78bdc64538df8725",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13445",
"triggerID" : "e6aae42cf093d4f44b93b79a78bdc64538df8725",
"triggerType" : "PUSH"
}, {
"hash" : "cccca8191e6800961467c726695c8ef7575d157a",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13449",
"triggerID" : "cccca8191e6800961467c726695c8ef7575d157a",
"triggerType" : "PUSH"
}, {
"hash" : "44ee00afbd5ab0c81c53ed5787e7c0701b711e41",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "44ee00afbd5ab0c81c53ed5787e7c0701b711e41",
"triggerType" : "PUSH"
}, {
"hash" : "8298e7dcbbd9809429dd4c2d4148ff5e46bdb43b",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13499",
"triggerID" : "8298e7dcbbd9809429dd4c2d4148ff5e46bdb43b",
"triggerType" : "PUSH"
}, {
"hash" : "d77a6c96e6616c2485ed787b7a1f0fedea97511b",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d77a6c96e6616c2485ed787b7a1f0fedea97511b",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1b43a411d22af19f12f14a79c1094c8dc5cfeb42 UNKNOWN
* 44ee00afbd5ab0c81c53ed5787e7c0701b711e41 UNKNOWN
* 8298e7dcbbd9809429dd4c2d4148ff5e46bdb43b Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13499)
* d77a6c96e6616c2485ed787b7a1f0fedea97511b UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] rmetzger commented on a change in pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
rmetzger commented on a change in pull request #14950:
URL: https://github.com/apache/flink/pull/14950#discussion_r577636469
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraph.java
##########
@@ -0,0 +1,1454 @@
+/*
+ * 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.executiongraph;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.ArchivedExecutionConfig;
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.api.common.accumulators.Accumulator;
+import org.apache.flink.api.common.accumulators.AccumulatorHelper;
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.SimpleCounter;
+import org.apache.flink.runtime.JobException;
+import org.apache.flink.runtime.accumulators.AccumulatorSnapshot;
+import org.apache.flink.runtime.accumulators.StringifiedAccumulatorResult;
+import org.apache.flink.runtime.blob.BlobWriter;
+import org.apache.flink.runtime.blob.PermanentBlobKey;
+import org.apache.flink.runtime.checkpoint.CheckpointCoordinator;
+import org.apache.flink.runtime.checkpoint.CheckpointFailureManager;
+import org.apache.flink.runtime.checkpoint.CheckpointIDCounter;
+import org.apache.flink.runtime.checkpoint.CheckpointPlanCalculator;
+import org.apache.flink.runtime.checkpoint.CheckpointStatsSnapshot;
+import org.apache.flink.runtime.checkpoint.CheckpointStatsTracker;
+import org.apache.flink.runtime.checkpoint.CheckpointsCleaner;
+import org.apache.flink.runtime.checkpoint.CompletedCheckpointStore;
+import org.apache.flink.runtime.checkpoint.ExecutionAttemptMappingProvider;
+import org.apache.flink.runtime.checkpoint.MasterTriggerRestoreHook;
+import org.apache.flink.runtime.checkpoint.OperatorCoordinatorCheckpointContext;
+import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor;
+import org.apache.flink.runtime.concurrent.FutureUtils;
+import org.apache.flink.runtime.concurrent.FutureUtils.ConjunctFuture;
+import org.apache.flink.runtime.concurrent.ScheduledExecutorServiceAdapter;
+import org.apache.flink.runtime.entrypoint.ClusterEntryPointExceptionUtils;
+import org.apache.flink.runtime.execution.ExecutionState;
+import org.apache.flink.runtime.executiongraph.failover.flip1.ResultPartitionAvailabilityChecker;
+import org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease.PartitionReleaseStrategy;
+import org.apache.flink.runtime.io.network.partition.JobMasterPartitionTracker;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
+import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
+import org.apache.flink.runtime.jobgraph.IntermediateResultPartitionID;
+import org.apache.flink.runtime.jobgraph.JobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+import org.apache.flink.runtime.jobgraph.ScheduleMode;
+import org.apache.flink.runtime.jobgraph.tasks.CheckpointCoordinatorConfiguration;
+import org.apache.flink.runtime.query.KvStateLocationRegistry;
+import org.apache.flink.runtime.scheduler.InternalFailuresListener;
+import org.apache.flink.runtime.scheduler.adapter.DefaultExecutionTopology;
+import org.apache.flink.runtime.scheduler.strategy.ExecutionVertexID;
+import org.apache.flink.runtime.scheduler.strategy.SchedulingExecutionVertex;
+import org.apache.flink.runtime.scheduler.strategy.SchedulingResultPartition;
+import org.apache.flink.runtime.scheduler.strategy.SchedulingTopology;
+import org.apache.flink.runtime.shuffle.ShuffleMaster;
+import org.apache.flink.runtime.state.CheckpointStorage;
+import org.apache.flink.runtime.state.SharedStateRegistry;
+import org.apache.flink.runtime.state.StateBackend;
+import org.apache.flink.runtime.taskmanager.DispatcherThreadFactory;
+import org.apache.flink.types.Either;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.FlinkException;
+import org.apache.flink.util.OptionalFailure;
+import org.apache.flink.util.SerializedThrowable;
+import org.apache.flink.util.SerializedValue;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkState;
+
+/** Default implementation of the {@link ExecutionGraph}. */
+public class DefaultExecutionGraph implements ExecutionGraph {
Review comment:
Yes
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14950:
URL: https://github.com/apache/flink/pull/14950#issuecomment-780431068
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13398",
"triggerID" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* e44c9cdb2e60872179635616e8ffb184bb0337b9 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13398)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] rmetzger commented on pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
rmetzger commented on pull request #14950:
URL: https://github.com/apache/flink/pull/14950#issuecomment-789560040
Thanks for your review! I rebased the PR and addressed all your comments.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] zentol commented on a change in pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
zentol commented on a change in pull request #14950:
URL: https://github.com/apache/flink/pull/14950#discussion_r579000369
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
##########
@@ -1021,203 +220,20 @@ public ArchivedExecutionConfig getArchivedExecutionConfig() {
*
* @return Termination future of this {@link ExecutionGraph}.
*/
- public CompletableFuture<JobStatus> getTerminationFuture() {
- return terminationFuture;
- }
+ CompletableFuture<JobStatus> getTerminationFuture();
@VisibleForTesting
- public JobStatus waitUntilTerminal() throws InterruptedException {
- try {
- return terminationFuture.get();
- } catch (ExecutionException e) {
- // this should never happen
- // it would be a bug, so we don't expect this to be handled and throw
- // an unchecked exception here
- throw new RuntimeException(e);
- }
- }
-
- // ------------------------------------------------------------------------
- // State Transitions
- // ------------------------------------------------------------------------
-
- public boolean transitionState(JobStatus current, JobStatus newState) {
- return transitionState(current, newState, null);
- }
-
- private void transitionState(JobStatus newState, Throwable error) {
- transitionState(state, newState, error);
- }
+ JobStatus waitUntilTerminal() throws InterruptedException;
- private boolean transitionState(JobStatus current, JobStatus newState, Throwable error) {
- assertRunningInJobMasterMainThread();
- // consistency check
- if (current.isTerminalState()) {
- String message = "Job is trying to leave terminal state " + current;
- LOG.error(message);
- throw new IllegalStateException(message);
- }
+ boolean transitionState(JobStatus current, JobStatus newState);
- // now do the actual state transition
- if (state == current) {
- state = newState;
- LOG.info(
- "Job {} ({}) switched from state {} to {}.",
- getJobName(),
- getJobID(),
- current,
- newState,
- error);
+ void incrementRestarts();
- stateTimestamps[newState.ordinal()] = System.currentTimeMillis();
- notifyJobStatusChange(newState, error);
- return true;
- } else {
- return false;
- }
- }
+ void initFailureCause(Throwable t);
- public void incrementRestarts() {
- numberOfRestartsCounter.inc();
- }
+ void vertexFinished();
Review comment:
That is only necessary because Executing works directly against ExecutionJobVertices/ExecutionVertices.
If there would be 2 small interfaces exposing the vertices the the JobVertex and deploy/markFailed for the ExecutionVertex, then you wouldn't need to mock these vertices in the test and would no longer require package-private APIs.
As it stands you're exposing a whole bunch of package-private methods to the outside which is just not acceptable.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] rmetzger commented on a change in pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
rmetzger commented on a change in pull request #14950:
URL: https://github.com/apache/flink/pull/14950#discussion_r578985387
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/declarative/StateTrackingMockExecutionGraph.java
##########
@@ -0,0 +1,459 @@
+/*
+ * 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.declarative;
+
+import org.apache.flink.api.common.ArchivedExecutionConfig;
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.api.common.accumulators.Accumulator;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.JobException;
+import org.apache.flink.runtime.accumulators.AccumulatorSnapshot;
+import org.apache.flink.runtime.accumulators.StringifiedAccumulatorResult;
+import org.apache.flink.runtime.blob.BlobWriter;
+import org.apache.flink.runtime.blob.PermanentBlobKey;
+import org.apache.flink.runtime.checkpoint.CheckpointCoordinator;
+import org.apache.flink.runtime.checkpoint.CheckpointIDCounter;
+import org.apache.flink.runtime.checkpoint.CheckpointStatsSnapshot;
+import org.apache.flink.runtime.checkpoint.CheckpointStatsTracker;
+import org.apache.flink.runtime.checkpoint.CheckpointsCleaner;
+import org.apache.flink.runtime.checkpoint.CompletedCheckpointStore;
+import org.apache.flink.runtime.checkpoint.MasterTriggerRestoreHook;
+import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor;
+import org.apache.flink.runtime.execution.ExecutionState;
+import org.apache.flink.runtime.executiongraph.ErrorInfo;
+import org.apache.flink.runtime.executiongraph.Execution;
+import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
+import org.apache.flink.runtime.executiongraph.ExecutionDeploymentListener;
+import org.apache.flink.runtime.executiongraph.ExecutionGraph;
+import org.apache.flink.runtime.executiongraph.ExecutionJobVertex;
+import org.apache.flink.runtime.executiongraph.ExecutionVertex;
+import org.apache.flink.runtime.executiongraph.IntermediateResult;
+import org.apache.flink.runtime.executiongraph.JobInformation;
+import org.apache.flink.runtime.executiongraph.JobStatusListener;
+import org.apache.flink.runtime.executiongraph.TaskExecutionStateTransition;
+import org.apache.flink.runtime.executiongraph.failover.flip1.ResultPartitionAvailabilityChecker;
+import org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease.PartitionReleaseStrategy;
+import org.apache.flink.runtime.io.network.partition.JobMasterPartitionTracker;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
+import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
+import org.apache.flink.runtime.jobgraph.JobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+import org.apache.flink.runtime.jobgraph.ScheduleMode;
+import org.apache.flink.runtime.jobgraph.tasks.CheckpointCoordinatorConfiguration;
+import org.apache.flink.runtime.query.KvStateLocationRegistry;
+import org.apache.flink.runtime.scheduler.InternalFailuresListener;
+import org.apache.flink.runtime.scheduler.strategy.SchedulingTopology;
+import org.apache.flink.runtime.shuffle.ShuffleMaster;
+import org.apache.flink.runtime.state.CheckpointStorage;
+import org.apache.flink.runtime.state.StateBackend;
+import org.apache.flink.types.Either;
+import org.apache.flink.util.OptionalFailure;
+import org.apache.flink.util.SerializedValue;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ForkJoinPool;
+
+/**
+ * Mocked ExecutionGraph which (partially) tracks the job status, and provides some basic mocks to
+ * create an {@link org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph} from this
+ * ExecutionGraph.
+ */
+class StateTrackingMockExecutionGraph implements ExecutionGraph {
+ private JobStatus state = JobStatus.INITIALIZING;
+ private final CompletableFuture<JobStatus> terminationFuture = new CompletableFuture<>();
+ private final JobID jobId = new JobID();
+ private final ArchivedExecutionConfig archivedExecutionConfig = new ExecutionConfig().archive();
+
+ // ---- methods to control the mock
+
+ void completeTerminationFuture(JobStatus finalStatus) {
+ terminationFuture.complete(finalStatus);
+ state = finalStatus;
+ }
+
+ // ---- interface implementations
+
+ @Override
+ public boolean updateState(TaskExecutionStateTransition state) {
+ return true;
+ }
+
+ @Override
+ public JobStatus getState() {
+ return state;
+ }
+
+ @Override
+ public CompletableFuture<JobStatus> getTerminationFuture() {
+ return terminationFuture;
+ }
+
+ @Override
+ public Executor getFutureExecutor() {
+ return ForkJoinPool.commonPool();
+ }
+
+ @Override
+ public void registerExecution(Execution exec) {
+ // no-op
+ }
+
+ @Override
+ public void cancel() {
+ state = JobStatus.CANCELLING;
+ }
+
+ @Override
+ public void failJob(Throwable cause) {
+ state = JobStatus.FAILING;
+ }
+
+ @Override
+ public void suspend(Throwable suspensionCause) {
+ state = JobStatus.SUSPENDED;
+ }
+
+ @Override
+ public void transitionToRunning() {
+ state = JobStatus.RUNNING;
+ }
+
+ // --- interface implementations: methods for creating an archived execution graph
+
+ @Override
+ public int getTotalNumberOfVertices() {
+ return 0;
+ }
+
+ @Override
+ public Iterable<ExecutionJobVertex> getVerticesTopologically() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public Map<JobVertexID, ExecutionJobVertex> getAllVertices() {
+ return Collections.emptyMap();
+ }
+
+ @Override
+ public Map<String, SerializedValue<OptionalFailure<Object>>> getAccumulatorsSerialized() {
+ return Collections.emptyMap();
+ }
+
+ @Override
+ public long getStatusTimestamp(JobStatus status) {
+ return 0L;
+ }
+
+ @Override
+ public String getJsonPlan() {
+ return "";
+ }
+
+ @Override
+ public JobID getJobID() {
+ return jobId;
+ }
+
+ @Override
+ public String getJobName() {
+ return "testJob";
+ }
+
+ @Nullable
+ @Override
+ public ErrorInfo getFailureInfo() {
+ return null;
+ }
+
+ @Nullable
+ @Override
+ public ArchivedExecutionConfig getArchivedExecutionConfig() {
+ return archivedExecutionConfig;
+ }
+
+ @Override
+ public boolean isStoppable() {
+ return false;
+ }
+
+ @Nullable
+ @Override
+ public CheckpointCoordinatorConfiguration getCheckpointCoordinatorConfiguration() {
+ return null;
+ }
+
+ @Nullable
+ @Override
+ public CheckpointStatsSnapshot getCheckpointStatsSnapshot() {
+ return null;
+ }
+
+ @Override
+ public Optional<String> getStateBackendName() {
+ return Optional.empty();
+ }
+
+ @Override
+ public Optional<String> getCheckpointStorageName() {
+ return Optional.empty();
+ }
+
+ @Override
+ public StringifiedAccumulatorResult[] getAccumulatorResultsStringified() {
+ return new StringifiedAccumulatorResult[0];
+ }
+
+ // -- remaining interface implementations: all unsupported
+
+ @Override
+ public boolean isArchived() {
+ throw new UnsupportedOperationException();
Review comment:
You mean adding a default method returning false in the `ExecutionGraph` interface? or just always returning false in the mock?
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] rmetzger commented on pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
rmetzger commented on pull request #14950:
URL: https://github.com/apache/flink/pull/14950#issuecomment-781149460
This PR is ready to review again!
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14950:
URL: https://github.com/apache/flink/pull/14950#issuecomment-780431068
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13398",
"triggerID" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"triggerType" : "PUSH"
}, {
"hash" : "bee61356ebeb5f99c96ce3193b3bd8205e61e44d",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13409",
"triggerID" : "bee61356ebeb5f99c96ce3193b3bd8205e61e44d",
"triggerType" : "PUSH"
}, {
"hash" : "1b43a411d22af19f12f14a79c1094c8dc5cfeb42",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "1b43a411d22af19f12f14a79c1094c8dc5cfeb42",
"triggerType" : "PUSH"
}, {
"hash" : "e6aae42cf093d4f44b93b79a78bdc64538df8725",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13445",
"triggerID" : "e6aae42cf093d4f44b93b79a78bdc64538df8725",
"triggerType" : "PUSH"
}, {
"hash" : "cccca8191e6800961467c726695c8ef7575d157a",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13449",
"triggerID" : "cccca8191e6800961467c726695c8ef7575d157a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* bee61356ebeb5f99c96ce3193b3bd8205e61e44d Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13409)
* 1b43a411d22af19f12f14a79c1094c8dc5cfeb42 UNKNOWN
* e6aae42cf093d4f44b93b79a78bdc64538df8725 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13445)
* cccca8191e6800961467c726695c8ef7575d157a Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13449)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14950:
URL: https://github.com/apache/flink/pull/14950#issuecomment-780431068
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13398",
"triggerID" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"triggerType" : "PUSH"
}, {
"hash" : "bee61356ebeb5f99c96ce3193b3bd8205e61e44d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13409",
"triggerID" : "bee61356ebeb5f99c96ce3193b3bd8205e61e44d",
"triggerType" : "PUSH"
}, {
"hash" : "1b43a411d22af19f12f14a79c1094c8dc5cfeb42",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "1b43a411d22af19f12f14a79c1094c8dc5cfeb42",
"triggerType" : "PUSH"
}, {
"hash" : "e6aae42cf093d4f44b93b79a78bdc64538df8725",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13445",
"triggerID" : "e6aae42cf093d4f44b93b79a78bdc64538df8725",
"triggerType" : "PUSH"
}, {
"hash" : "cccca8191e6800961467c726695c8ef7575d157a",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13449",
"triggerID" : "cccca8191e6800961467c726695c8ef7575d157a",
"triggerType" : "PUSH"
}, {
"hash" : "44ee00afbd5ab0c81c53ed5787e7c0701b711e41",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "44ee00afbd5ab0c81c53ed5787e7c0701b711e41",
"triggerType" : "PUSH"
}, {
"hash" : "8298e7dcbbd9809429dd4c2d4148ff5e46bdb43b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13499",
"triggerID" : "8298e7dcbbd9809429dd4c2d4148ff5e46bdb43b",
"triggerType" : "PUSH"
}, {
"hash" : "d77a6c96e6616c2485ed787b7a1f0fedea97511b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13599",
"triggerID" : "d77a6c96e6616c2485ed787b7a1f0fedea97511b",
"triggerType" : "PUSH"
}, {
"hash" : "c1274ae69962c108df47f9939b3454154724acbb",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14049",
"triggerID" : "c1274ae69962c108df47f9939b3454154724acbb",
"triggerType" : "PUSH"
}, {
"hash" : "03ac0474397f9055837dc5727d634ac42ee18ce7",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "03ac0474397f9055837dc5727d634ac42ee18ce7",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1b43a411d22af19f12f14a79c1094c8dc5cfeb42 UNKNOWN
* 44ee00afbd5ab0c81c53ed5787e7c0701b711e41 UNKNOWN
* c1274ae69962c108df47f9939b3454154724acbb Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14049)
* 03ac0474397f9055837dc5727d634ac42ee18ce7 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] rmetzger closed pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
rmetzger closed pull request #14950:
URL: https://github.com/apache/flink/pull/14950
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] rmetzger commented on pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
rmetzger commented on pull request #14950:
URL: https://github.com/apache/flink/pull/14950#issuecomment-789780217
Thanks for approving the change. I filed a ticket for the CI instability: https://issues.apache.org/jira/browse/FLINK-21587
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] zentol commented on a change in pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
zentol commented on a change in pull request #14950:
URL: https://github.com/apache/flink/pull/14950#discussion_r580963986
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
##########
@@ -1021,203 +220,20 @@ public ArchivedExecutionConfig getArchivedExecutionConfig() {
*
* @return Termination future of this {@link ExecutionGraph}.
*/
- public CompletableFuture<JobStatus> getTerminationFuture() {
- return terminationFuture;
- }
+ CompletableFuture<JobStatus> getTerminationFuture();
@VisibleForTesting
- public JobStatus waitUntilTerminal() throws InterruptedException {
- try {
- return terminationFuture.get();
- } catch (ExecutionException e) {
- // this should never happen
- // it would be a bug, so we don't expect this to be handled and throw
- // an unchecked exception here
- throw new RuntimeException(e);
- }
- }
-
- // ------------------------------------------------------------------------
- // State Transitions
- // ------------------------------------------------------------------------
-
- public boolean transitionState(JobStatus current, JobStatus newState) {
- return transitionState(current, newState, null);
- }
-
- private void transitionState(JobStatus newState, Throwable error) {
- transitionState(state, newState, error);
- }
+ JobStatus waitUntilTerminal() throws InterruptedException;
- private boolean transitionState(JobStatus current, JobStatus newState, Throwable error) {
- assertRunningInJobMasterMainThread();
- // consistency check
- if (current.isTerminalState()) {
- String message = "Job is trying to leave terminal state " + current;
- LOG.error(message);
- throw new IllegalStateException(message);
- }
+ boolean transitionState(JobStatus current, JobStatus newState);
- // now do the actual state transition
- if (state == current) {
- state = newState;
- LOG.info(
- "Job {} ({}) switched from state {} to {}.",
- getJobName(),
- getJobID(),
- current,
- newState,
- error);
+ void incrementRestarts();
- stateTimestamps[newState.ordinal()] = System.currentTimeMillis();
- notifyJobStatusChange(newState, error);
- return true;
- } else {
- return false;
- }
- }
+ void initFailureCause(Throwable t);
- public void incrementRestarts() {
- numberOfRestartsCounter.inc();
- }
+ void vertexFinished();
Review comment:
Yes I didn't explain it very well, and I don't think my suggestion (according to the phrasing) as is would work particularly well.
The original issue we wanted to solve with this PR is that mocking an EG is annoying and error-prone, so we introduced an interface to remedy that.
However, because all the Execution* classes are pretty much one intermingled mess, and the Executing state works against the Execution(Job)Vertex, you ended up still having a dependency on the package-private APIs, which you also exposed via the interface.
You them introduced an interface for the EG containing the internal APIs; this allows you to mock the internal API of the EG that the vertex requires. While it works it doesn't really address the core issue of not having a clear API. Despite only wanting to assert that the state deploys/cancels some vertices we end up having to mock EG-internal APIs and all that jazz.
Alternatively, and this is what I was more going for, we could go with a similar approach like what you did for the EG for the remaining Execution* classes. In that model you'd have to mock a lot less methods, and can entirely ignore how these classes internally communicate.
I would think that this wouldn't be soooo much overhead, in particular because we need a lot less methods from these classes; so we can probably get by with a tiny extension of the Access* interfaces.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14950:
URL: https://github.com/apache/flink/pull/14950#issuecomment-780431068
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13398",
"triggerID" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"triggerType" : "PUSH"
}, {
"hash" : "bee61356ebeb5f99c96ce3193b3bd8205e61e44d",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13409",
"triggerID" : "bee61356ebeb5f99c96ce3193b3bd8205e61e44d",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* e44c9cdb2e60872179635616e8ffb184bb0337b9 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13398)
* bee61356ebeb5f99c96ce3193b3bd8205e61e44d Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13409)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14950:
URL: https://github.com/apache/flink/pull/14950#issuecomment-780431068
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13398",
"triggerID" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"triggerType" : "PUSH"
}, {
"hash" : "bee61356ebeb5f99c96ce3193b3bd8205e61e44d",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13409",
"triggerID" : "bee61356ebeb5f99c96ce3193b3bd8205e61e44d",
"triggerType" : "PUSH"
}, {
"hash" : "1b43a411d22af19f12f14a79c1094c8dc5cfeb42",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "1b43a411d22af19f12f14a79c1094c8dc5cfeb42",
"triggerType" : "PUSH"
}, {
"hash" : "e6aae42cf093d4f44b93b79a78bdc64538df8725",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13445",
"triggerID" : "e6aae42cf093d4f44b93b79a78bdc64538df8725",
"triggerType" : "PUSH"
}, {
"hash" : "cccca8191e6800961467c726695c8ef7575d157a",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "cccca8191e6800961467c726695c8ef7575d157a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* bee61356ebeb5f99c96ce3193b3bd8205e61e44d Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13409)
* 1b43a411d22af19f12f14a79c1094c8dc5cfeb42 UNKNOWN
* e6aae42cf093d4f44b93b79a78bdc64538df8725 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13445)
* cccca8191e6800961467c726695c8ef7575d157a UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14950:
URL: https://github.com/apache/flink/pull/14950#issuecomment-780431068
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13398",
"triggerID" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"triggerType" : "PUSH"
}, {
"hash" : "bee61356ebeb5f99c96ce3193b3bd8205e61e44d",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13409",
"triggerID" : "bee61356ebeb5f99c96ce3193b3bd8205e61e44d",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* bee61356ebeb5f99c96ce3193b3bd8205e61e44d Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13409)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] rmetzger commented on a change in pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
rmetzger commented on a change in pull request #14950:
URL: https://github.com/apache/flink/pull/14950#discussion_r578998929
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/declarative/StateTrackingMockExecutionGraph.java
##########
@@ -0,0 +1,459 @@
+/*
+ * 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.declarative;
+
+import org.apache.flink.api.common.ArchivedExecutionConfig;
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.api.common.accumulators.Accumulator;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.JobException;
+import org.apache.flink.runtime.accumulators.AccumulatorSnapshot;
+import org.apache.flink.runtime.accumulators.StringifiedAccumulatorResult;
+import org.apache.flink.runtime.blob.BlobWriter;
+import org.apache.flink.runtime.blob.PermanentBlobKey;
+import org.apache.flink.runtime.checkpoint.CheckpointCoordinator;
+import org.apache.flink.runtime.checkpoint.CheckpointIDCounter;
+import org.apache.flink.runtime.checkpoint.CheckpointStatsSnapshot;
+import org.apache.flink.runtime.checkpoint.CheckpointStatsTracker;
+import org.apache.flink.runtime.checkpoint.CheckpointsCleaner;
+import org.apache.flink.runtime.checkpoint.CompletedCheckpointStore;
+import org.apache.flink.runtime.checkpoint.MasterTriggerRestoreHook;
+import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor;
+import org.apache.flink.runtime.execution.ExecutionState;
+import org.apache.flink.runtime.executiongraph.ErrorInfo;
+import org.apache.flink.runtime.executiongraph.Execution;
+import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
+import org.apache.flink.runtime.executiongraph.ExecutionDeploymentListener;
+import org.apache.flink.runtime.executiongraph.ExecutionGraph;
+import org.apache.flink.runtime.executiongraph.ExecutionJobVertex;
+import org.apache.flink.runtime.executiongraph.ExecutionVertex;
+import org.apache.flink.runtime.executiongraph.IntermediateResult;
+import org.apache.flink.runtime.executiongraph.JobInformation;
+import org.apache.flink.runtime.executiongraph.JobStatusListener;
+import org.apache.flink.runtime.executiongraph.TaskExecutionStateTransition;
+import org.apache.flink.runtime.executiongraph.failover.flip1.ResultPartitionAvailabilityChecker;
+import org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease.PartitionReleaseStrategy;
+import org.apache.flink.runtime.io.network.partition.JobMasterPartitionTracker;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
+import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
+import org.apache.flink.runtime.jobgraph.JobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+import org.apache.flink.runtime.jobgraph.ScheduleMode;
+import org.apache.flink.runtime.jobgraph.tasks.CheckpointCoordinatorConfiguration;
+import org.apache.flink.runtime.query.KvStateLocationRegistry;
+import org.apache.flink.runtime.scheduler.InternalFailuresListener;
+import org.apache.flink.runtime.scheduler.strategy.SchedulingTopology;
+import org.apache.flink.runtime.shuffle.ShuffleMaster;
+import org.apache.flink.runtime.state.CheckpointStorage;
+import org.apache.flink.runtime.state.StateBackend;
+import org.apache.flink.types.Either;
+import org.apache.flink.util.OptionalFailure;
+import org.apache.flink.util.SerializedValue;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ForkJoinPool;
+
+/**
+ * Mocked ExecutionGraph which (partially) tracks the job status, and provides some basic mocks to
+ * create an {@link org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph} from this
+ * ExecutionGraph.
+ */
+class StateTrackingMockExecutionGraph implements ExecutionGraph {
+ private JobStatus state = JobStatus.INITIALIZING;
+ private final CompletableFuture<JobStatus> terminationFuture = new CompletableFuture<>();
+ private final JobID jobId = new JobID();
+ private final ArchivedExecutionConfig archivedExecutionConfig = new ExecutionConfig().archive();
+
+ // ---- methods to control the mock
+
+ void completeTerminationFuture(JobStatus finalStatus) {
+ terminationFuture.complete(finalStatus);
+ state = finalStatus;
+ }
+
+ // ---- interface implementations
+
+ @Override
+ public boolean updateState(TaskExecutionStateTransition state) {
+ return true;
+ }
+
+ @Override
+ public JobStatus getState() {
+ return state;
+ }
+
+ @Override
+ public CompletableFuture<JobStatus> getTerminationFuture() {
+ return terminationFuture;
+ }
+
+ @Override
+ public Executor getFutureExecutor() {
+ return ForkJoinPool.commonPool();
+ }
+
+ @Override
+ public void registerExecution(Execution exec) {
+ // no-op
+ }
+
+ @Override
+ public void cancel() {
+ state = JobStatus.CANCELLING;
+ }
+
+ @Override
+ public void failJob(Throwable cause) {
+ state = JobStatus.FAILING;
+ }
+
+ @Override
+ public void suspend(Throwable suspensionCause) {
+ state = JobStatus.SUSPENDED;
+ }
+
+ @Override
+ public void transitionToRunning() {
+ state = JobStatus.RUNNING;
+ }
+
+ // --- interface implementations: methods for creating an archived execution graph
+
+ @Override
+ public int getTotalNumberOfVertices() {
+ return 0;
+ }
+
+ @Override
+ public Iterable<ExecutionJobVertex> getVerticesTopologically() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public Map<JobVertexID, ExecutionJobVertex> getAllVertices() {
+ return Collections.emptyMap();
+ }
+
+ @Override
+ public Map<String, SerializedValue<OptionalFailure<Object>>> getAccumulatorsSerialized() {
+ return Collections.emptyMap();
+ }
+
+ @Override
+ public long getStatusTimestamp(JobStatus status) {
+ return 0L;
+ }
+
+ @Override
+ public String getJsonPlan() {
+ return "";
+ }
+
+ @Override
+ public JobID getJobID() {
+ return jobId;
+ }
+
+ @Override
+ public String getJobName() {
+ return "testJob";
+ }
+
+ @Nullable
+ @Override
+ public ErrorInfo getFailureInfo() {
+ return null;
+ }
+
+ @Nullable
+ @Override
+ public ArchivedExecutionConfig getArchivedExecutionConfig() {
+ return archivedExecutionConfig;
+ }
+
+ @Override
+ public boolean isStoppable() {
+ return false;
+ }
+
+ @Nullable
+ @Override
+ public CheckpointCoordinatorConfiguration getCheckpointCoordinatorConfiguration() {
+ return null;
+ }
+
+ @Nullable
+ @Override
+ public CheckpointStatsSnapshot getCheckpointStatsSnapshot() {
+ return null;
+ }
+
+ @Override
+ public Optional<String> getStateBackendName() {
+ return Optional.empty();
+ }
+
+ @Override
+ public Optional<String> getCheckpointStorageName() {
+ return Optional.empty();
+ }
+
+ @Override
+ public StringifiedAccumulatorResult[] getAccumulatorResultsStringified() {
+ return new StringifiedAccumulatorResult[0];
+ }
+
+ // -- remaining interface implementations: all unsupported
+
+ @Override
+ public boolean isArchived() {
+ throw new UnsupportedOperationException();
Review comment:
Ah, good catch. Will remove it.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] zentol commented on a change in pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
zentol commented on a change in pull request #14950:
URL: https://github.com/apache/flink/pull/14950#discussion_r585421476
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/StateTrackingMockExecutionGraph.java
##########
@@ -0,0 +1,342 @@
+/*
+ * 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;
+
+import org.apache.flink.api.common.ArchivedExecutionConfig;
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.api.common.accumulators.Accumulator;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.JobException;
+import org.apache.flink.runtime.accumulators.AccumulatorSnapshot;
+import org.apache.flink.runtime.accumulators.StringifiedAccumulatorResult;
+import org.apache.flink.runtime.checkpoint.CheckpointCoordinator;
+import org.apache.flink.runtime.checkpoint.CheckpointIDCounter;
+import org.apache.flink.runtime.checkpoint.CheckpointStatsSnapshot;
+import org.apache.flink.runtime.checkpoint.CheckpointStatsTracker;
+import org.apache.flink.runtime.checkpoint.CheckpointsCleaner;
+import org.apache.flink.runtime.checkpoint.CompletedCheckpointStore;
+import org.apache.flink.runtime.checkpoint.MasterTriggerRestoreHook;
+import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor;
+import org.apache.flink.runtime.executiongraph.ErrorInfo;
+import org.apache.flink.runtime.executiongraph.Execution;
+import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
+import org.apache.flink.runtime.executiongraph.ExecutionGraph;
+import org.apache.flink.runtime.executiongraph.ExecutionJobVertex;
+import org.apache.flink.runtime.executiongraph.ExecutionVertex;
+import org.apache.flink.runtime.executiongraph.IntermediateResult;
+import org.apache.flink.runtime.executiongraph.JobStatusListener;
+import org.apache.flink.runtime.executiongraph.TaskExecutionStateTransition;
+import org.apache.flink.runtime.executiongraph.failover.flip1.ResultPartitionAvailabilityChecker;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
+import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
+import org.apache.flink.runtime.jobgraph.JobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+import org.apache.flink.runtime.jobgraph.tasks.CheckpointCoordinatorConfiguration;
+import org.apache.flink.runtime.query.KvStateLocationRegistry;
+import org.apache.flink.runtime.scheduler.InternalFailuresListener;
+import org.apache.flink.runtime.scheduler.strategy.SchedulingTopology;
+import org.apache.flink.runtime.state.CheckpointStorage;
+import org.apache.flink.runtime.state.StateBackend;
+import org.apache.flink.util.OptionalFailure;
+import org.apache.flink.util.SerializedValue;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Mocked ExecutionGraph which (partially) tracks the job status, and provides some basic mocks to
+ * create an {@link org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph} from this
+ * ExecutionGraph.
+ */
+class StateTrackingMockExecutionGraph implements ExecutionGraph {
+ private JobStatus state = JobStatus.INITIALIZING;
+ private final CompletableFuture<JobStatus> terminationFuture = new CompletableFuture<>();
+ private final JobID jobId = new JobID();
+ private final ArchivedExecutionConfig archivedExecutionConfig = new ExecutionConfig().archive();
Review comment:
this one could be static
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/RestartingTest.java
##########
@@ -148,8 +134,8 @@ public Restarting createRestartingState(
public Restarting createRestartingState(MockRestartingContext ctx)
throws JobException, JobExecutionException {
- ExecutionGraph executionGraph = TestingExecutionGraphBuilder.newBuilder().build();
- return createRestartingState(ctx, executionGraph);
+ // ExecutionGraph executionGraph = TestingDefaultExecutionGraphBuilder.newBuilder().build();
Review comment:
.
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphDeploymentWithBlobServerTest.java
##########
@@ -87,7 +87,7 @@ public void shutdownBlobServer() throws IOException {
}
@Override
- protected void checkJobOffloaded(ExecutionGraph eg) throws Exception {
+ protected void checkJobOffloaded(DefaultExecutionGraph eg) throws Exception {
Review comment:
this seems incorrect?
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionVertex.java
##########
@@ -309,7 +313,7 @@ public AllocationID getLatestPriorAllocation() {
}
}
- public ExecutionGraph getExecutionGraph() {
+ public InternalExecutionGraphAccessor getExecutionAccessor() {
Review comment:
```suggestion
public InternalExecutionGraphAccessor getExecutionGraphAccessor() {
```
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphDeploymentWithBlobServerTest.java
##########
@@ -87,7 +87,7 @@ public void shutdownBlobServer() throws IOException {
}
@Override
- protected void checkJobOffloaded(ExecutionGraph eg) throws Exception {
+ protected void checkJobOffloaded(DefaultExecutionGraph eg) throws Exception {
Review comment:
Oh, `getJobInformationOrBlobKey` is not part of the EG interface... :/
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphRescalingTest.java
##########
@@ -58,7 +58,8 @@ public void testExecutionGraphArbitraryDopConstructionTest() throws Exception {
createVerticesForSimpleBipartiteJobGraph(initialParallelism, maxParallelism);
final JobGraph jobGraph = new JobGraph(jobVertices);
- ExecutionGraph eg = TestingExecutionGraphBuilder.newBuilder().setJobGraph(jobGraph).build();
Review comment:
should this test (and others) be renamed to refer to the `DefaultExecutionGraph`?
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] zentol commented on a change in pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
zentol commented on a change in pull request #14950:
URL: https://github.com/apache/flink/pull/14950#discussion_r578977899
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/declarative/StateTrackingMockExecutionGraph.java
##########
@@ -0,0 +1,459 @@
+/*
+ * 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.declarative;
+
+import org.apache.flink.api.common.ArchivedExecutionConfig;
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.api.common.accumulators.Accumulator;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.JobException;
+import org.apache.flink.runtime.accumulators.AccumulatorSnapshot;
+import org.apache.flink.runtime.accumulators.StringifiedAccumulatorResult;
+import org.apache.flink.runtime.blob.BlobWriter;
+import org.apache.flink.runtime.blob.PermanentBlobKey;
+import org.apache.flink.runtime.checkpoint.CheckpointCoordinator;
+import org.apache.flink.runtime.checkpoint.CheckpointIDCounter;
+import org.apache.flink.runtime.checkpoint.CheckpointStatsSnapshot;
+import org.apache.flink.runtime.checkpoint.CheckpointStatsTracker;
+import org.apache.flink.runtime.checkpoint.CheckpointsCleaner;
+import org.apache.flink.runtime.checkpoint.CompletedCheckpointStore;
+import org.apache.flink.runtime.checkpoint.MasterTriggerRestoreHook;
+import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor;
+import org.apache.flink.runtime.execution.ExecutionState;
+import org.apache.flink.runtime.executiongraph.ErrorInfo;
+import org.apache.flink.runtime.executiongraph.Execution;
+import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
+import org.apache.flink.runtime.executiongraph.ExecutionDeploymentListener;
+import org.apache.flink.runtime.executiongraph.ExecutionGraph;
+import org.apache.flink.runtime.executiongraph.ExecutionJobVertex;
+import org.apache.flink.runtime.executiongraph.ExecutionVertex;
+import org.apache.flink.runtime.executiongraph.IntermediateResult;
+import org.apache.flink.runtime.executiongraph.JobInformation;
+import org.apache.flink.runtime.executiongraph.JobStatusListener;
+import org.apache.flink.runtime.executiongraph.TaskExecutionStateTransition;
+import org.apache.flink.runtime.executiongraph.failover.flip1.ResultPartitionAvailabilityChecker;
+import org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease.PartitionReleaseStrategy;
+import org.apache.flink.runtime.io.network.partition.JobMasterPartitionTracker;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
+import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
+import org.apache.flink.runtime.jobgraph.JobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+import org.apache.flink.runtime.jobgraph.ScheduleMode;
+import org.apache.flink.runtime.jobgraph.tasks.CheckpointCoordinatorConfiguration;
+import org.apache.flink.runtime.query.KvStateLocationRegistry;
+import org.apache.flink.runtime.scheduler.InternalFailuresListener;
+import org.apache.flink.runtime.scheduler.strategy.SchedulingTopology;
+import org.apache.flink.runtime.shuffle.ShuffleMaster;
+import org.apache.flink.runtime.state.CheckpointStorage;
+import org.apache.flink.runtime.state.StateBackend;
+import org.apache.flink.types.Either;
+import org.apache.flink.util.OptionalFailure;
+import org.apache.flink.util.SerializedValue;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ForkJoinPool;
+
+/**
+ * Mocked ExecutionGraph which (partially) tracks the job status, and provides some basic mocks to
+ * create an {@link org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph} from this
+ * ExecutionGraph.
+ */
+class StateTrackingMockExecutionGraph implements ExecutionGraph {
+ private JobStatus state = JobStatus.INITIALIZING;
+ private final CompletableFuture<JobStatus> terminationFuture = new CompletableFuture<>();
+ private final JobID jobId = new JobID();
+ private final ArchivedExecutionConfig archivedExecutionConfig = new ExecutionConfig().archive();
+
+ // ---- methods to control the mock
+
+ void completeTerminationFuture(JobStatus finalStatus) {
+ terminationFuture.complete(finalStatus);
+ state = finalStatus;
+ }
+
+ // ---- interface implementations
+
+ @Override
+ public boolean updateState(TaskExecutionStateTransition state) {
+ return true;
+ }
+
+ @Override
+ public JobStatus getState() {
+ return state;
+ }
+
+ @Override
+ public CompletableFuture<JobStatus> getTerminationFuture() {
+ return terminationFuture;
+ }
+
+ @Override
+ public Executor getFutureExecutor() {
+ return ForkJoinPool.commonPool();
+ }
+
+ @Override
+ public void registerExecution(Execution exec) {
+ // no-op
+ }
+
+ @Override
+ public void cancel() {
+ state = JobStatus.CANCELLING;
+ }
+
+ @Override
+ public void failJob(Throwable cause) {
+ state = JobStatus.FAILING;
+ }
+
+ @Override
+ public void suspend(Throwable suspensionCause) {
+ state = JobStatus.SUSPENDED;
+ }
+
+ @Override
+ public void transitionToRunning() {
+ state = JobStatus.RUNNING;
+ }
+
+ // --- interface implementations: methods for creating an archived execution graph
+
+ @Override
+ public int getTotalNumberOfVertices() {
+ return 0;
+ }
+
+ @Override
+ public Iterable<ExecutionJobVertex> getVerticesTopologically() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public Map<JobVertexID, ExecutionJobVertex> getAllVertices() {
+ return Collections.emptyMap();
+ }
+
+ @Override
+ public Map<String, SerializedValue<OptionalFailure<Object>>> getAccumulatorsSerialized() {
+ return Collections.emptyMap();
+ }
+
+ @Override
+ public long getStatusTimestamp(JobStatus status) {
+ return 0L;
+ }
+
+ @Override
+ public String getJsonPlan() {
+ return "";
+ }
+
+ @Override
+ public JobID getJobID() {
+ return jobId;
+ }
+
+ @Override
+ public String getJobName() {
+ return "testJob";
+ }
+
+ @Nullable
+ @Override
+ public ErrorInfo getFailureInfo() {
+ return null;
+ }
+
+ @Nullable
+ @Override
+ public ArchivedExecutionConfig getArchivedExecutionConfig() {
+ return archivedExecutionConfig;
+ }
+
+ @Override
+ public boolean isStoppable() {
+ return false;
+ }
+
+ @Nullable
+ @Override
+ public CheckpointCoordinatorConfiguration getCheckpointCoordinatorConfiguration() {
+ return null;
+ }
+
+ @Nullable
+ @Override
+ public CheckpointStatsSnapshot getCheckpointStatsSnapshot() {
+ return null;
+ }
+
+ @Override
+ public Optional<String> getStateBackendName() {
+ return Optional.empty();
+ }
+
+ @Override
+ public Optional<String> getCheckpointStorageName() {
+ return Optional.empty();
+ }
+
+ @Override
+ public StringifiedAccumulatorResult[] getAccumulatorResultsStringified() {
+ return new StringifiedAccumulatorResult[0];
+ }
+
+ // -- remaining interface implementations: all unsupported
+
+ @Override
+ public boolean isArchived() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void start(@Nonnull ComponentMainThreadExecutor jobMasterMainThreadExecutor) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public int getNumberOfExecutionJobVertices() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public SchedulingTopology getSchedulingTopology() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public ScheduleMode getScheduleMode() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Nonnull
+ @Override
+ public ComponentMainThreadExecutor getJobMasterMainThreadExecutor() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void enableCheckpointing(
+ CheckpointCoordinatorConfiguration chkConfig,
+ List<MasterTriggerRestoreHook<?>> masterHooks,
+ CheckpointIDCounter checkpointIDCounter,
+ CompletedCheckpointStore checkpointStore,
+ StateBackend checkpointStateBackend,
+ CheckpointStorage checkpointStorage,
+ CheckpointStatsTracker statsTracker,
+ CheckpointsCleaner checkpointsCleaner) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Nullable
+ @Override
+ public CheckpointCoordinator getCheckpointCoordinator() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public KvStateLocationRegistry getKvStateLocationRegistry() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void setJsonPlan(String jsonPlan) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Either<SerializedValue<JobInformation>, PermanentBlobKey> getJobInformationOrBlobKey() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Configuration getJobConfiguration() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public ClassLoader getUserClassLoader() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Throwable getFailureCause() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Iterable<ExecutionVertex> getAllExecutionVertices() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public ExecutionJobVertex getJobVertex(JobVertexID id) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public long getNumberOfRestarts() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Map<IntermediateDataSetID, IntermediateResult> getAllIntermediateResults() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public BlobWriter getBlobWriter() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Map<String, OptionalFailure<Accumulator<?, ?>>> aggregateUserAccumulators() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void setInternalTaskFailuresListener(
+ InternalFailuresListener internalTaskFailuresListener) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void attachJobGraph(List<JobVertex> topologiallySorted) throws JobException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void failGlobal(Throwable t) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public JobStatus waitUntilTerminal() throws InterruptedException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean transitionState(JobStatus current, JobStatus newState) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void incrementRestarts() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void initFailureCause(Throwable t) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void vertexFinished() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void vertexUnFinished() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void notifyPartitionDataAvailable(ResultPartitionID partitionId) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Map<ExecutionAttemptID, Execution> getRegisteredExecutions() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateAccumulators(AccumulatorSnapshot accumulatorSnapshot) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void registerJobStatusListener(JobStatusListener listener) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void assertRunningInJobMasterMainThread() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void notifySchedulerNgAboutInternalTaskFailure(
+ ExecutionAttemptID attemptId,
+ Throwable t,
+ boolean cancelTask,
+ boolean releasePartitions) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public ShuffleMaster<?> getShuffleMaster() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public JobMasterPartitionTracker getPartitionTracker() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public ResultPartitionAvailabilityChecker getResultPartitionAvailabilityChecker() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public PartitionReleaseStrategy getPartitionReleaseStrategy() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public ExecutionDeploymentListener getExecutionDeploymentListener() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void deregisterExecution(Execution exec) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void notifyExecutionChange(Execution execution, ExecutionState newExecutionState) {
+ throw new UnsupportedOperationException();
Review comment:
given that updateTaskExecutionState is a safe operation, I'd say this one should be too.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] rmetzger commented on a change in pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
rmetzger commented on a change in pull request #14950:
URL: https://github.com/apache/flink/pull/14950#discussion_r585302773
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
##########
@@ -1021,203 +220,20 @@ public ArchivedExecutionConfig getArchivedExecutionConfig() {
*
* @return Termination future of this {@link ExecutionGraph}.
*/
- public CompletableFuture<JobStatus> getTerminationFuture() {
- return terminationFuture;
- }
+ CompletableFuture<JobStatus> getTerminationFuture();
@VisibleForTesting
- public JobStatus waitUntilTerminal() throws InterruptedException {
- try {
- return terminationFuture.get();
- } catch (ExecutionException e) {
- // this should never happen
- // it would be a bug, so we don't expect this to be handled and throw
- // an unchecked exception here
- throw new RuntimeException(e);
- }
- }
-
- // ------------------------------------------------------------------------
- // State Transitions
- // ------------------------------------------------------------------------
-
- public boolean transitionState(JobStatus current, JobStatus newState) {
- return transitionState(current, newState, null);
- }
-
- private void transitionState(JobStatus newState, Throwable error) {
- transitionState(state, newState, error);
- }
+ JobStatus waitUntilTerminal() throws InterruptedException;
- private boolean transitionState(JobStatus current, JobStatus newState, Throwable error) {
- assertRunningInJobMasterMainThread();
- // consistency check
- if (current.isTerminalState()) {
- String message = "Job is trying to leave terminal state " + current;
- LOG.error(message);
- throw new IllegalStateException(message);
- }
+ boolean transitionState(JobStatus current, JobStatus newState);
- // now do the actual state transition
- if (state == current) {
- state = newState;
- LOG.info(
- "Job {} ({}) switched from state {} to {}.",
- getJobName(),
- getJobID(),
- current,
- newState,
- error);
+ void incrementRestarts();
- stateTimestamps[newState.ordinal()] = System.currentTimeMillis();
- notifyJobStatusChange(newState, error);
- return true;
- } else {
- return false;
- }
- }
+ void initFailureCause(Throwable t);
- public void incrementRestarts() {
- numberOfRestartsCounter.inc();
- }
+ void vertexFinished();
Review comment:
Thanks a lot for your explanation. Now I understand what you mean.
How would you feel about merging this PR as-is, because it already improves the situation a lot, and we address the introduction of interfaces for the remaining Execution* classes in a follow up?
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14950:
URL: https://github.com/apache/flink/pull/14950#issuecomment-780431068
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13398",
"triggerID" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"triggerType" : "PUSH"
}, {
"hash" : "bee61356ebeb5f99c96ce3193b3bd8205e61e44d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13409",
"triggerID" : "bee61356ebeb5f99c96ce3193b3bd8205e61e44d",
"triggerType" : "PUSH"
}, {
"hash" : "1b43a411d22af19f12f14a79c1094c8dc5cfeb42",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "1b43a411d22af19f12f14a79c1094c8dc5cfeb42",
"triggerType" : "PUSH"
}, {
"hash" : "e6aae42cf093d4f44b93b79a78bdc64538df8725",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13445",
"triggerID" : "e6aae42cf093d4f44b93b79a78bdc64538df8725",
"triggerType" : "PUSH"
}, {
"hash" : "cccca8191e6800961467c726695c8ef7575d157a",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13449",
"triggerID" : "cccca8191e6800961467c726695c8ef7575d157a",
"triggerType" : "PUSH"
}, {
"hash" : "44ee00afbd5ab0c81c53ed5787e7c0701b711e41",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "44ee00afbd5ab0c81c53ed5787e7c0701b711e41",
"triggerType" : "PUSH"
}, {
"hash" : "8298e7dcbbd9809429dd4c2d4148ff5e46bdb43b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13499",
"triggerID" : "8298e7dcbbd9809429dd4c2d4148ff5e46bdb43b",
"triggerType" : "PUSH"
}, {
"hash" : "d77a6c96e6616c2485ed787b7a1f0fedea97511b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13599",
"triggerID" : "d77a6c96e6616c2485ed787b7a1f0fedea97511b",
"triggerType" : "PUSH"
}, {
"hash" : "c1274ae69962c108df47f9939b3454154724acbb",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14049",
"triggerID" : "c1274ae69962c108df47f9939b3454154724acbb",
"triggerType" : "PUSH"
}, {
"hash" : "03ac0474397f9055837dc5727d634ac42ee18ce7",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14071",
"triggerID" : "03ac0474397f9055837dc5727d634ac42ee18ce7",
"triggerType" : "PUSH"
}, {
"hash" : "42a6b627410d26ff3c1447a82c139e2e17faa707",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "42a6b627410d26ff3c1447a82c139e2e17faa707",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1b43a411d22af19f12f14a79c1094c8dc5cfeb42 UNKNOWN
* 44ee00afbd5ab0c81c53ed5787e7c0701b711e41 UNKNOWN
* 03ac0474397f9055837dc5727d634ac42ee18ce7 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14071)
* 42a6b627410d26ff3c1447a82c139e2e17faa707 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14950:
URL: https://github.com/apache/flink/pull/14950#issuecomment-780431068
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13398",
"triggerID" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"triggerType" : "PUSH"
}, {
"hash" : "bee61356ebeb5f99c96ce3193b3bd8205e61e44d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13409",
"triggerID" : "bee61356ebeb5f99c96ce3193b3bd8205e61e44d",
"triggerType" : "PUSH"
}, {
"hash" : "1b43a411d22af19f12f14a79c1094c8dc5cfeb42",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "1b43a411d22af19f12f14a79c1094c8dc5cfeb42",
"triggerType" : "PUSH"
}, {
"hash" : "e6aae42cf093d4f44b93b79a78bdc64538df8725",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13445",
"triggerID" : "e6aae42cf093d4f44b93b79a78bdc64538df8725",
"triggerType" : "PUSH"
}, {
"hash" : "cccca8191e6800961467c726695c8ef7575d157a",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13449",
"triggerID" : "cccca8191e6800961467c726695c8ef7575d157a",
"triggerType" : "PUSH"
}, {
"hash" : "44ee00afbd5ab0c81c53ed5787e7c0701b711e41",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "44ee00afbd5ab0c81c53ed5787e7c0701b711e41",
"triggerType" : "PUSH"
}, {
"hash" : "8298e7dcbbd9809429dd4c2d4148ff5e46bdb43b",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13499",
"triggerID" : "8298e7dcbbd9809429dd4c2d4148ff5e46bdb43b",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1b43a411d22af19f12f14a79c1094c8dc5cfeb42 UNKNOWN
* 44ee00afbd5ab0c81c53ed5787e7c0701b711e41 UNKNOWN
* 8298e7dcbbd9809429dd4c2d4148ff5e46bdb43b Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13499)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14950:
URL: https://github.com/apache/flink/pull/14950#issuecomment-780431068
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13398",
"triggerID" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* e44c9cdb2e60872179635616e8ffb184bb0337b9 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13398)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] rmetzger commented on a change in pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
rmetzger commented on a change in pull request #14950:
URL: https://github.com/apache/flink/pull/14950#discussion_r577640124
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/MockExecutionGraphBase.java
##########
@@ -0,0 +1,424 @@
+/*
+ * 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.executiongraph;
+
+import org.apache.flink.api.common.ArchivedExecutionConfig;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.api.common.accumulators.Accumulator;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.JobException;
+import org.apache.flink.runtime.accumulators.AccumulatorSnapshot;
+import org.apache.flink.runtime.accumulators.StringifiedAccumulatorResult;
+import org.apache.flink.runtime.blob.BlobWriter;
+import org.apache.flink.runtime.blob.PermanentBlobKey;
+import org.apache.flink.runtime.checkpoint.CheckpointCoordinator;
+import org.apache.flink.runtime.checkpoint.CheckpointIDCounter;
+import org.apache.flink.runtime.checkpoint.CheckpointStatsSnapshot;
+import org.apache.flink.runtime.checkpoint.CheckpointStatsTracker;
+import org.apache.flink.runtime.checkpoint.CheckpointsCleaner;
+import org.apache.flink.runtime.checkpoint.CompletedCheckpointStore;
+import org.apache.flink.runtime.checkpoint.MasterTriggerRestoreHook;
+import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor;
+import org.apache.flink.runtime.execution.ExecutionState;
+import org.apache.flink.runtime.executiongraph.failover.flip1.ResultPartitionAvailabilityChecker;
+import org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease.PartitionReleaseStrategy;
+import org.apache.flink.runtime.io.network.partition.JobMasterPartitionTracker;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
+import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
+import org.apache.flink.runtime.jobgraph.JobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+import org.apache.flink.runtime.jobgraph.ScheduleMode;
+import org.apache.flink.runtime.jobgraph.tasks.CheckpointCoordinatorConfiguration;
+import org.apache.flink.runtime.query.KvStateLocationRegistry;
+import org.apache.flink.runtime.scheduler.InternalFailuresListener;
+import org.apache.flink.runtime.scheduler.strategy.SchedulingTopology;
+import org.apache.flink.runtime.shuffle.ShuffleMaster;
+import org.apache.flink.runtime.state.CheckpointStorage;
+import org.apache.flink.runtime.state.StateBackend;
+import org.apache.flink.types.Either;
+import org.apache.flink.util.OptionalFailure;
+import org.apache.flink.util.SerializedValue;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+
+/** Base class for mocking the ExecutionGraph. */
+public class MockExecutionGraphBase implements ExecutionGraph {
+
+ @Override
+ public String getJsonPlan() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public JobID getJobID() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public String getJobName() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public JobStatus getState() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Nullable
+ @Override
+ public ErrorInfo getFailureInfo() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public long getStatusTimestamp(JobStatus status) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Nullable
+ @Override
+ public CheckpointCoordinatorConfiguration getCheckpointCoordinatorConfiguration() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Nullable
+ @Override
+ public CheckpointStatsSnapshot getCheckpointStatsSnapshot() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Nullable
+ @Override
+ public ArchivedExecutionConfig getArchivedExecutionConfig() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean isStoppable() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public StringifiedAccumulatorResult[] getAccumulatorResultsStringified() {
+ return new StringifiedAccumulatorResult[0];
+ }
+
+ @Override
+ public Map<String, SerializedValue<OptionalFailure<Object>>> getAccumulatorsSerialized() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean isArchived() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Optional<String> getStateBackendName() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Optional<String> getCheckpointStorageName() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void start(@Nonnull ComponentMainThreadExecutor jobMasterMainThreadExecutor) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public int getNumberOfExecutionJobVertices() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public SchedulingTopology getSchedulingTopology() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public ScheduleMode getScheduleMode() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Nonnull
+ @Override
+ public ComponentMainThreadExecutor getJobMasterMainThreadExecutor() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void enableCheckpointing(
+ CheckpointCoordinatorConfiguration chkConfig,
+ List<MasterTriggerRestoreHook<?>> masterHooks,
+ CheckpointIDCounter checkpointIDCounter,
+ CompletedCheckpointStore checkpointStore,
+ StateBackend checkpointStateBackend,
+ CheckpointStorage checkpointStorage,
+ CheckpointStatsTracker statsTracker,
+ CheckpointsCleaner checkpointsCleaner) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Nullable
+ @Override
+ public CheckpointCoordinator getCheckpointCoordinator() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public KvStateLocationRegistry getKvStateLocationRegistry() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void setJsonPlan(String jsonPlan) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Either<SerializedValue<JobInformation>, PermanentBlobKey> getJobInformationOrBlobKey() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Configuration getJobConfiguration() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public ClassLoader getUserClassLoader() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Throwable getFailureCause() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Iterable<ExecutionJobVertex> getVerticesTopologically() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Iterable<ExecutionVertex> getAllExecutionVertices() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public ExecutionJobVertex getJobVertex(JobVertexID id) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Map<JobVertexID, ExecutionJobVertex> getAllVertices() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public long getNumberOfRestarts() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public int getTotalNumberOfVertices() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Map<IntermediateDataSetID, IntermediateResult> getAllIntermediateResults() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public BlobWriter getBlobWriter() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Executor getFutureExecutor() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Map<String, OptionalFailure<Accumulator<?, ?>>> aggregateUserAccumulators() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void setInternalTaskFailuresListener(
+ InternalFailuresListener internalTaskFailuresListener) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void attachJobGraph(List<JobVertex> topologiallySorted) throws JobException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void transitionToRunning() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void cancel() {
+ throw new UnsupportedOperationException();
Review comment:
For my use-case, I'd almost say the opposite is true. I consider it useful to see right away what I've missed to implement in my mock by having an exception thrown. Making cancel a no-op would require me to first understand why the cancellation hasn't happened as expected. With an exception, I notice it right away.
The MockExecutionGraphBase is really just a utility for the state tests. I agree that it might be of debatable use for others. Would moving the MockBase to the declarative scheduler tests package resolve your concern?
I can understand that in other test scenarios, people want other mocks, but they are free to set them up.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] rmetzger commented on a change in pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
rmetzger commented on a change in pull request #14950:
URL: https://github.com/apache/flink/pull/14950#discussion_r578181797
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/MockExecutionGraphBase.java
##########
@@ -0,0 +1,424 @@
+/*
+ * 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.executiongraph;
+
+import org.apache.flink.api.common.ArchivedExecutionConfig;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.api.common.accumulators.Accumulator;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.JobException;
+import org.apache.flink.runtime.accumulators.AccumulatorSnapshot;
+import org.apache.flink.runtime.accumulators.StringifiedAccumulatorResult;
+import org.apache.flink.runtime.blob.BlobWriter;
+import org.apache.flink.runtime.blob.PermanentBlobKey;
+import org.apache.flink.runtime.checkpoint.CheckpointCoordinator;
+import org.apache.flink.runtime.checkpoint.CheckpointIDCounter;
+import org.apache.flink.runtime.checkpoint.CheckpointStatsSnapshot;
+import org.apache.flink.runtime.checkpoint.CheckpointStatsTracker;
+import org.apache.flink.runtime.checkpoint.CheckpointsCleaner;
+import org.apache.flink.runtime.checkpoint.CompletedCheckpointStore;
+import org.apache.flink.runtime.checkpoint.MasterTriggerRestoreHook;
+import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor;
+import org.apache.flink.runtime.execution.ExecutionState;
+import org.apache.flink.runtime.executiongraph.failover.flip1.ResultPartitionAvailabilityChecker;
+import org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease.PartitionReleaseStrategy;
+import org.apache.flink.runtime.io.network.partition.JobMasterPartitionTracker;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
+import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
+import org.apache.flink.runtime.jobgraph.JobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+import org.apache.flink.runtime.jobgraph.ScheduleMode;
+import org.apache.flink.runtime.jobgraph.tasks.CheckpointCoordinatorConfiguration;
+import org.apache.flink.runtime.query.KvStateLocationRegistry;
+import org.apache.flink.runtime.scheduler.InternalFailuresListener;
+import org.apache.flink.runtime.scheduler.strategy.SchedulingTopology;
+import org.apache.flink.runtime.shuffle.ShuffleMaster;
+import org.apache.flink.runtime.state.CheckpointStorage;
+import org.apache.flink.runtime.state.StateBackend;
+import org.apache.flink.types.Either;
+import org.apache.flink.util.OptionalFailure;
+import org.apache.flink.util.SerializedValue;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+
+/** Base class for mocking the ExecutionGraph. */
+public class MockExecutionGraphBase implements ExecutionGraph {
+
+ @Override
+ public String getJsonPlan() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public JobID getJobID() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public String getJobName() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public JobStatus getState() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Nullable
+ @Override
+ public ErrorInfo getFailureInfo() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public long getStatusTimestamp(JobStatus status) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Nullable
+ @Override
+ public CheckpointCoordinatorConfiguration getCheckpointCoordinatorConfiguration() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Nullable
+ @Override
+ public CheckpointStatsSnapshot getCheckpointStatsSnapshot() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Nullable
+ @Override
+ public ArchivedExecutionConfig getArchivedExecutionConfig() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean isStoppable() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public StringifiedAccumulatorResult[] getAccumulatorResultsStringified() {
+ return new StringifiedAccumulatorResult[0];
+ }
+
+ @Override
+ public Map<String, SerializedValue<OptionalFailure<Object>>> getAccumulatorsSerialized() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean isArchived() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Optional<String> getStateBackendName() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Optional<String> getCheckpointStorageName() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void start(@Nonnull ComponentMainThreadExecutor jobMasterMainThreadExecutor) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public int getNumberOfExecutionJobVertices() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public SchedulingTopology getSchedulingTopology() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public ScheduleMode getScheduleMode() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Nonnull
+ @Override
+ public ComponentMainThreadExecutor getJobMasterMainThreadExecutor() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void enableCheckpointing(
+ CheckpointCoordinatorConfiguration chkConfig,
+ List<MasterTriggerRestoreHook<?>> masterHooks,
+ CheckpointIDCounter checkpointIDCounter,
+ CompletedCheckpointStore checkpointStore,
+ StateBackend checkpointStateBackend,
+ CheckpointStorage checkpointStorage,
+ CheckpointStatsTracker statsTracker,
+ CheckpointsCleaner checkpointsCleaner) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Nullable
+ @Override
+ public CheckpointCoordinator getCheckpointCoordinator() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public KvStateLocationRegistry getKvStateLocationRegistry() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void setJsonPlan(String jsonPlan) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Either<SerializedValue<JobInformation>, PermanentBlobKey> getJobInformationOrBlobKey() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Configuration getJobConfiguration() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public ClassLoader getUserClassLoader() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Throwable getFailureCause() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Iterable<ExecutionJobVertex> getVerticesTopologically() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Iterable<ExecutionVertex> getAllExecutionVertices() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public ExecutionJobVertex getJobVertex(JobVertexID id) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Map<JobVertexID, ExecutionJobVertex> getAllVertices() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public long getNumberOfRestarts() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public int getTotalNumberOfVertices() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Map<IntermediateDataSetID, IntermediateResult> getAllIntermediateResults() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public BlobWriter getBlobWriter() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Executor getFutureExecutor() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Map<String, OptionalFailure<Accumulator<?, ?>>> aggregateUserAccumulators() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void setInternalTaskFailuresListener(
+ InternalFailuresListener internalTaskFailuresListener) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void attachJobGraph(List<JobVertex> topologiallySorted) throws JobException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void transitionToRunning() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void cancel() {
+ throw new UnsupportedOperationException();
Review comment:
After working on fixing the failing tests, I realize that I need to revisit my MockExecutionGraphBase strategy.
I'll move it to the declarative scheduler package and customize it more for my needs.
I'll ping you when I need another review.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14950:
URL: https://github.com/apache/flink/pull/14950#issuecomment-780431068
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13398",
"triggerID" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"triggerType" : "PUSH"
}, {
"hash" : "bee61356ebeb5f99c96ce3193b3bd8205e61e44d",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "bee61356ebeb5f99c96ce3193b3bd8205e61e44d",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* e44c9cdb2e60872179635616e8ffb184bb0337b9 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13398)
* bee61356ebeb5f99c96ce3193b3bd8205e61e44d UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] zentol commented on a change in pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
zentol commented on a change in pull request #14950:
URL: https://github.com/apache/flink/pull/14950#discussion_r577520012
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/declarative/ExecutingTest.java
##########
@@ -466,49 +455,48 @@ public Throwable getFailureCause() {
}
}
- private static class MockExecutionGraph extends ExecutionGraph {
+ private static class MockExecutionGraph extends MockExecutionGraphBase {
private final boolean updateStateReturnValue;
private final Supplier<Iterable<ExecutionJobVertex>> getVerticesTopologicallySupplier;
+ private JobStatus state = JobStatus.INITIALIZING;
+ private CompletableFuture<JobStatus> terminationFuture = new CompletableFuture<>();
- MockExecutionGraph(Supplier<Iterable<ExecutionJobVertex>> getVerticesTopologicallySupplier)
- throws IOException {
+ MockExecutionGraph(
+ Supplier<Iterable<ExecutionJobVertex>> getVerticesTopologicallySupplier) {
this(false, getVerticesTopologicallySupplier);
}
- MockExecutionGraph(boolean updateStateReturnValue) throws IOException {
+ MockExecutionGraph(boolean updateStateReturnValue) {
this(updateStateReturnValue, null);
}
private MockExecutionGraph(
boolean updateStateReturnValue,
- Supplier<Iterable<ExecutionJobVertex>> getVerticesTopologicallySupplier)
- throws IOException {
- super(
- new JobInformation(
- new JobID(),
- "Test Job",
- new SerializedValue<>(new ExecutionConfig()),
- new Configuration(),
- Collections.emptyList(),
- Collections.emptyList()),
- TestingUtils.defaultExecutor(),
- TestingUtils.defaultExecutor(),
- AkkaUtils.getDefaultTimeout(),
- 1,
- ExecutionGraph.class.getClassLoader(),
- VoidBlobWriter.getInstance(),
- PartitionReleaseStrategyFactoryLoader.loadPartitionReleaseStrategyFactory(
- new Configuration()),
- NettyShuffleMaster.INSTANCE,
- NoOpJobMasterPartitionTracker.INSTANCE,
- ScheduleMode.EAGER,
- NoOpExecutionDeploymentListener.get(),
- (execution, newState) -> {},
- 0L);
+ Supplier<Iterable<ExecutionJobVertex>> getVerticesTopologicallySupplier) {
this.updateStateReturnValue = updateStateReturnValue;
this.getVerticesTopologicallySupplier = getVerticesTopologicallySupplier;
}
+ @Override
+ public void transitionToRunning() {
+ state = JobStatus.RUNNING;
+ }
+
+ @Override
+ public JobStatus getState() {
+ return state;
+ }
+
+ @Override
+ public JobID getJobID() {
+ return new JobID();
Review comment:
this could lead to subtle issues if `getJobID` is called multiple times.
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/MockExecutionGraphBase.java
##########
@@ -0,0 +1,424 @@
+/*
+ * 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.executiongraph;
+
+import org.apache.flink.api.common.ArchivedExecutionConfig;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.api.common.accumulators.Accumulator;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.JobException;
+import org.apache.flink.runtime.accumulators.AccumulatorSnapshot;
+import org.apache.flink.runtime.accumulators.StringifiedAccumulatorResult;
+import org.apache.flink.runtime.blob.BlobWriter;
+import org.apache.flink.runtime.blob.PermanentBlobKey;
+import org.apache.flink.runtime.checkpoint.CheckpointCoordinator;
+import org.apache.flink.runtime.checkpoint.CheckpointIDCounter;
+import org.apache.flink.runtime.checkpoint.CheckpointStatsSnapshot;
+import org.apache.flink.runtime.checkpoint.CheckpointStatsTracker;
+import org.apache.flink.runtime.checkpoint.CheckpointsCleaner;
+import org.apache.flink.runtime.checkpoint.CompletedCheckpointStore;
+import org.apache.flink.runtime.checkpoint.MasterTriggerRestoreHook;
+import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor;
+import org.apache.flink.runtime.execution.ExecutionState;
+import org.apache.flink.runtime.executiongraph.failover.flip1.ResultPartitionAvailabilityChecker;
+import org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease.PartitionReleaseStrategy;
+import org.apache.flink.runtime.io.network.partition.JobMasterPartitionTracker;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
+import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
+import org.apache.flink.runtime.jobgraph.JobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+import org.apache.flink.runtime.jobgraph.ScheduleMode;
+import org.apache.flink.runtime.jobgraph.tasks.CheckpointCoordinatorConfiguration;
+import org.apache.flink.runtime.query.KvStateLocationRegistry;
+import org.apache.flink.runtime.scheduler.InternalFailuresListener;
+import org.apache.flink.runtime.scheduler.strategy.SchedulingTopology;
+import org.apache.flink.runtime.shuffle.ShuffleMaster;
+import org.apache.flink.runtime.state.CheckpointStorage;
+import org.apache.flink.runtime.state.StateBackend;
+import org.apache.flink.types.Either;
+import org.apache.flink.util.OptionalFailure;
+import org.apache.flink.util.SerializedValue;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+
+/** Base class for mocking the ExecutionGraph. */
+public class MockExecutionGraphBase implements ExecutionGraph {
+
+ @Override
+ public String getJsonPlan() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public JobID getJobID() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public String getJobName() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public JobStatus getState() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Nullable
+ @Override
+ public ErrorInfo getFailureInfo() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public long getStatusTimestamp(JobStatus status) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Nullable
+ @Override
+ public CheckpointCoordinatorConfiguration getCheckpointCoordinatorConfiguration() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Nullable
+ @Override
+ public CheckpointStatsSnapshot getCheckpointStatsSnapshot() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Nullable
+ @Override
+ public ArchivedExecutionConfig getArchivedExecutionConfig() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean isStoppable() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public StringifiedAccumulatorResult[] getAccumulatorResultsStringified() {
+ return new StringifiedAccumulatorResult[0];
+ }
+
+ @Override
+ public Map<String, SerializedValue<OptionalFailure<Object>>> getAccumulatorsSerialized() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean isArchived() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Optional<String> getStateBackendName() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Optional<String> getCheckpointStorageName() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void start(@Nonnull ComponentMainThreadExecutor jobMasterMainThreadExecutor) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public int getNumberOfExecutionJobVertices() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public SchedulingTopology getSchedulingTopology() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public ScheduleMode getScheduleMode() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Nonnull
+ @Override
+ public ComponentMainThreadExecutor getJobMasterMainThreadExecutor() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void enableCheckpointing(
+ CheckpointCoordinatorConfiguration chkConfig,
+ List<MasterTriggerRestoreHook<?>> masterHooks,
+ CheckpointIDCounter checkpointIDCounter,
+ CompletedCheckpointStore checkpointStore,
+ StateBackend checkpointStateBackend,
+ CheckpointStorage checkpointStorage,
+ CheckpointStatsTracker statsTracker,
+ CheckpointsCleaner checkpointsCleaner) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Nullable
+ @Override
+ public CheckpointCoordinator getCheckpointCoordinator() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public KvStateLocationRegistry getKvStateLocationRegistry() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void setJsonPlan(String jsonPlan) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Either<SerializedValue<JobInformation>, PermanentBlobKey> getJobInformationOrBlobKey() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Configuration getJobConfiguration() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public ClassLoader getUserClassLoader() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Throwable getFailureCause() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Iterable<ExecutionJobVertex> getVerticesTopologically() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Iterable<ExecutionVertex> getAllExecutionVertices() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public ExecutionJobVertex getJobVertex(JobVertexID id) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Map<JobVertexID, ExecutionJobVertex> getAllVertices() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public long getNumberOfRestarts() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public int getTotalNumberOfVertices() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Map<IntermediateDataSetID, IntermediateResult> getAllIntermediateResults() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public BlobWriter getBlobWriter() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Executor getFutureExecutor() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Map<String, OptionalFailure<Accumulator<?, ?>>> aggregateUserAccumulators() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void setInternalTaskFailuresListener(
+ InternalFailuresListener internalTaskFailuresListener) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void attachJobGraph(List<JobVertex> topologiallySorted) throws JobException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void transitionToRunning() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void cancel() {
+ throw new UnsupportedOperationException();
Review comment:
Throwing `UnsupportedOperationExceptions` whenever any method that is not explicitly mocked by the test is called will make this quite annoying to use.
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/MockExecutionGraphBase.java
##########
@@ -0,0 +1,424 @@
+/*
+ * 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.executiongraph;
+
+import org.apache.flink.api.common.ArchivedExecutionConfig;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.api.common.accumulators.Accumulator;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.JobException;
+import org.apache.flink.runtime.accumulators.AccumulatorSnapshot;
+import org.apache.flink.runtime.accumulators.StringifiedAccumulatorResult;
+import org.apache.flink.runtime.blob.BlobWriter;
+import org.apache.flink.runtime.blob.PermanentBlobKey;
+import org.apache.flink.runtime.checkpoint.CheckpointCoordinator;
+import org.apache.flink.runtime.checkpoint.CheckpointIDCounter;
+import org.apache.flink.runtime.checkpoint.CheckpointStatsSnapshot;
+import org.apache.flink.runtime.checkpoint.CheckpointStatsTracker;
+import org.apache.flink.runtime.checkpoint.CheckpointsCleaner;
+import org.apache.flink.runtime.checkpoint.CompletedCheckpointStore;
+import org.apache.flink.runtime.checkpoint.MasterTriggerRestoreHook;
+import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor;
+import org.apache.flink.runtime.execution.ExecutionState;
+import org.apache.flink.runtime.executiongraph.failover.flip1.ResultPartitionAvailabilityChecker;
+import org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease.PartitionReleaseStrategy;
+import org.apache.flink.runtime.io.network.partition.JobMasterPartitionTracker;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
+import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
+import org.apache.flink.runtime.jobgraph.JobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+import org.apache.flink.runtime.jobgraph.ScheduleMode;
+import org.apache.flink.runtime.jobgraph.tasks.CheckpointCoordinatorConfiguration;
+import org.apache.flink.runtime.query.KvStateLocationRegistry;
+import org.apache.flink.runtime.scheduler.InternalFailuresListener;
+import org.apache.flink.runtime.scheduler.strategy.SchedulingTopology;
+import org.apache.flink.runtime.shuffle.ShuffleMaster;
+import org.apache.flink.runtime.state.CheckpointStorage;
+import org.apache.flink.runtime.state.StateBackend;
+import org.apache.flink.types.Either;
+import org.apache.flink.util.OptionalFailure;
+import org.apache.flink.util.SerializedValue;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+
+/** Base class for mocking the ExecutionGraph. */
+public class MockExecutionGraphBase implements ExecutionGraph {
+
+ @Override
+ public String getJsonPlan() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public JobID getJobID() {
+ throw new UnsupportedOperationException();
Review comment:
I feel like at least for some of these methods we could provide reasonable defaults.
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraph.java
##########
@@ -0,0 +1,1454 @@
+/*
+ * 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.executiongraph;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.ArchivedExecutionConfig;
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.api.common.accumulators.Accumulator;
+import org.apache.flink.api.common.accumulators.AccumulatorHelper;
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.SimpleCounter;
+import org.apache.flink.runtime.JobException;
+import org.apache.flink.runtime.accumulators.AccumulatorSnapshot;
+import org.apache.flink.runtime.accumulators.StringifiedAccumulatorResult;
+import org.apache.flink.runtime.blob.BlobWriter;
+import org.apache.flink.runtime.blob.PermanentBlobKey;
+import org.apache.flink.runtime.checkpoint.CheckpointCoordinator;
+import org.apache.flink.runtime.checkpoint.CheckpointFailureManager;
+import org.apache.flink.runtime.checkpoint.CheckpointIDCounter;
+import org.apache.flink.runtime.checkpoint.CheckpointPlanCalculator;
+import org.apache.flink.runtime.checkpoint.CheckpointStatsSnapshot;
+import org.apache.flink.runtime.checkpoint.CheckpointStatsTracker;
+import org.apache.flink.runtime.checkpoint.CheckpointsCleaner;
+import org.apache.flink.runtime.checkpoint.CompletedCheckpointStore;
+import org.apache.flink.runtime.checkpoint.ExecutionAttemptMappingProvider;
+import org.apache.flink.runtime.checkpoint.MasterTriggerRestoreHook;
+import org.apache.flink.runtime.checkpoint.OperatorCoordinatorCheckpointContext;
+import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor;
+import org.apache.flink.runtime.concurrent.FutureUtils;
+import org.apache.flink.runtime.concurrent.FutureUtils.ConjunctFuture;
+import org.apache.flink.runtime.concurrent.ScheduledExecutorServiceAdapter;
+import org.apache.flink.runtime.entrypoint.ClusterEntryPointExceptionUtils;
+import org.apache.flink.runtime.execution.ExecutionState;
+import org.apache.flink.runtime.executiongraph.failover.flip1.ResultPartitionAvailabilityChecker;
+import org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease.PartitionReleaseStrategy;
+import org.apache.flink.runtime.io.network.partition.JobMasterPartitionTracker;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
+import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
+import org.apache.flink.runtime.jobgraph.IntermediateResultPartitionID;
+import org.apache.flink.runtime.jobgraph.JobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+import org.apache.flink.runtime.jobgraph.ScheduleMode;
+import org.apache.flink.runtime.jobgraph.tasks.CheckpointCoordinatorConfiguration;
+import org.apache.flink.runtime.query.KvStateLocationRegistry;
+import org.apache.flink.runtime.scheduler.InternalFailuresListener;
+import org.apache.flink.runtime.scheduler.adapter.DefaultExecutionTopology;
+import org.apache.flink.runtime.scheduler.strategy.ExecutionVertexID;
+import org.apache.flink.runtime.scheduler.strategy.SchedulingExecutionVertex;
+import org.apache.flink.runtime.scheduler.strategy.SchedulingResultPartition;
+import org.apache.flink.runtime.scheduler.strategy.SchedulingTopology;
+import org.apache.flink.runtime.shuffle.ShuffleMaster;
+import org.apache.flink.runtime.state.CheckpointStorage;
+import org.apache.flink.runtime.state.SharedStateRegistry;
+import org.apache.flink.runtime.state.StateBackend;
+import org.apache.flink.runtime.taskmanager.DispatcherThreadFactory;
+import org.apache.flink.types.Either;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.FlinkException;
+import org.apache.flink.util.OptionalFailure;
+import org.apache.flink.util.SerializedThrowable;
+import org.apache.flink.util.SerializedValue;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkState;
+
+/** Default implementation of the {@link ExecutionGraph}. */
+public class DefaultExecutionGraph implements ExecutionGraph {
Review comment:
This is just a copy, right?
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14950:
URL: https://github.com/apache/flink/pull/14950#issuecomment-780431068
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13398",
"triggerID" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"triggerType" : "PUSH"
}, {
"hash" : "bee61356ebeb5f99c96ce3193b3bd8205e61e44d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13409",
"triggerID" : "bee61356ebeb5f99c96ce3193b3bd8205e61e44d",
"triggerType" : "PUSH"
}, {
"hash" : "1b43a411d22af19f12f14a79c1094c8dc5cfeb42",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "1b43a411d22af19f12f14a79c1094c8dc5cfeb42",
"triggerType" : "PUSH"
}, {
"hash" : "e6aae42cf093d4f44b93b79a78bdc64538df8725",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13445",
"triggerID" : "e6aae42cf093d4f44b93b79a78bdc64538df8725",
"triggerType" : "PUSH"
}, {
"hash" : "cccca8191e6800961467c726695c8ef7575d157a",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13449",
"triggerID" : "cccca8191e6800961467c726695c8ef7575d157a",
"triggerType" : "PUSH"
}, {
"hash" : "44ee00afbd5ab0c81c53ed5787e7c0701b711e41",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "44ee00afbd5ab0c81c53ed5787e7c0701b711e41",
"triggerType" : "PUSH"
}, {
"hash" : "8298e7dcbbd9809429dd4c2d4148ff5e46bdb43b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13499",
"triggerID" : "8298e7dcbbd9809429dd4c2d4148ff5e46bdb43b",
"triggerType" : "PUSH"
}, {
"hash" : "d77a6c96e6616c2485ed787b7a1f0fedea97511b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13599",
"triggerID" : "d77a6c96e6616c2485ed787b7a1f0fedea97511b",
"triggerType" : "PUSH"
}, {
"hash" : "c1274ae69962c108df47f9939b3454154724acbb",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14049",
"triggerID" : "c1274ae69962c108df47f9939b3454154724acbb",
"triggerType" : "PUSH"
}, {
"hash" : "03ac0474397f9055837dc5727d634ac42ee18ce7",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14071",
"triggerID" : "03ac0474397f9055837dc5727d634ac42ee18ce7",
"triggerType" : "PUSH"
}, {
"hash" : "42a6b627410d26ff3c1447a82c139e2e17faa707",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14135",
"triggerID" : "42a6b627410d26ff3c1447a82c139e2e17faa707",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1b43a411d22af19f12f14a79c1094c8dc5cfeb42 UNKNOWN
* 44ee00afbd5ab0c81c53ed5787e7c0701b711e41 UNKNOWN
* 03ac0474397f9055837dc5727d634ac42ee18ce7 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14071)
* 42a6b627410d26ff3c1447a82c139e2e17faa707 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14135)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] rmetzger commented on a change in pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
rmetzger commented on a change in pull request #14950:
URL: https://github.com/apache/flink/pull/14950#discussion_r578216296
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraph.java
##########
@@ -0,0 +1,1454 @@
+/*
+ * 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.executiongraph;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.ArchivedExecutionConfig;
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.api.common.accumulators.Accumulator;
+import org.apache.flink.api.common.accumulators.AccumulatorHelper;
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.SimpleCounter;
+import org.apache.flink.runtime.JobException;
+import org.apache.flink.runtime.accumulators.AccumulatorSnapshot;
+import org.apache.flink.runtime.accumulators.StringifiedAccumulatorResult;
+import org.apache.flink.runtime.blob.BlobWriter;
+import org.apache.flink.runtime.blob.PermanentBlobKey;
+import org.apache.flink.runtime.checkpoint.CheckpointCoordinator;
+import org.apache.flink.runtime.checkpoint.CheckpointFailureManager;
+import org.apache.flink.runtime.checkpoint.CheckpointIDCounter;
+import org.apache.flink.runtime.checkpoint.CheckpointPlanCalculator;
+import org.apache.flink.runtime.checkpoint.CheckpointStatsSnapshot;
+import org.apache.flink.runtime.checkpoint.CheckpointStatsTracker;
+import org.apache.flink.runtime.checkpoint.CheckpointsCleaner;
+import org.apache.flink.runtime.checkpoint.CompletedCheckpointStore;
+import org.apache.flink.runtime.checkpoint.ExecutionAttemptMappingProvider;
+import org.apache.flink.runtime.checkpoint.MasterTriggerRestoreHook;
+import org.apache.flink.runtime.checkpoint.OperatorCoordinatorCheckpointContext;
+import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor;
+import org.apache.flink.runtime.concurrent.FutureUtils;
+import org.apache.flink.runtime.concurrent.FutureUtils.ConjunctFuture;
+import org.apache.flink.runtime.concurrent.ScheduledExecutorServiceAdapter;
+import org.apache.flink.runtime.entrypoint.ClusterEntryPointExceptionUtils;
+import org.apache.flink.runtime.execution.ExecutionState;
+import org.apache.flink.runtime.executiongraph.failover.flip1.ResultPartitionAvailabilityChecker;
+import org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease.PartitionReleaseStrategy;
+import org.apache.flink.runtime.io.network.partition.JobMasterPartitionTracker;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
+import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
+import org.apache.flink.runtime.jobgraph.IntermediateResultPartitionID;
+import org.apache.flink.runtime.jobgraph.JobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+import org.apache.flink.runtime.jobgraph.ScheduleMode;
+import org.apache.flink.runtime.jobgraph.tasks.CheckpointCoordinatorConfiguration;
+import org.apache.flink.runtime.query.KvStateLocationRegistry;
+import org.apache.flink.runtime.scheduler.InternalFailuresListener;
+import org.apache.flink.runtime.scheduler.adapter.DefaultExecutionTopology;
+import org.apache.flink.runtime.scheduler.strategy.ExecutionVertexID;
+import org.apache.flink.runtime.scheduler.strategy.SchedulingExecutionVertex;
+import org.apache.flink.runtime.scheduler.strategy.SchedulingResultPartition;
+import org.apache.flink.runtime.scheduler.strategy.SchedulingTopology;
+import org.apache.flink.runtime.shuffle.ShuffleMaster;
+import org.apache.flink.runtime.state.CheckpointStorage;
+import org.apache.flink.runtime.state.SharedStateRegistry;
+import org.apache.flink.runtime.state.StateBackend;
+import org.apache.flink.runtime.taskmanager.DispatcherThreadFactory;
+import org.apache.flink.types.Either;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.FlinkException;
+import org.apache.flink.util.OptionalFailure;
+import org.apache.flink.util.SerializedThrowable;
+import org.apache.flink.util.SerializedValue;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkState;
+
+/** Default implementation of the {@link ExecutionGraph}. */
+public class DefaultExecutionGraph implements ExecutionGraph {
+
+ /** The log object used for debugging. */
+ static final Logger LOG = LoggerFactory.getLogger(ExecutionGraph.class);
Review comment:
I changed this from DefaultExecutionGraph to ExecutionGraph
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14950:
URL: https://github.com/apache/flink/pull/14950#issuecomment-780431068
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13398",
"triggerID" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"triggerType" : "PUSH"
}, {
"hash" : "bee61356ebeb5f99c96ce3193b3bd8205e61e44d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13409",
"triggerID" : "bee61356ebeb5f99c96ce3193b3bd8205e61e44d",
"triggerType" : "PUSH"
}, {
"hash" : "1b43a411d22af19f12f14a79c1094c8dc5cfeb42",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "1b43a411d22af19f12f14a79c1094c8dc5cfeb42",
"triggerType" : "PUSH"
}, {
"hash" : "e6aae42cf093d4f44b93b79a78bdc64538df8725",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13445",
"triggerID" : "e6aae42cf093d4f44b93b79a78bdc64538df8725",
"triggerType" : "PUSH"
}, {
"hash" : "cccca8191e6800961467c726695c8ef7575d157a",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13449",
"triggerID" : "cccca8191e6800961467c726695c8ef7575d157a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1b43a411d22af19f12f14a79c1094c8dc5cfeb42 UNKNOWN
* e6aae42cf093d4f44b93b79a78bdc64538df8725 Azure: [CANCELED](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13445)
* cccca8191e6800961467c726695c8ef7575d157a Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13449)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14950:
URL: https://github.com/apache/flink/pull/14950#issuecomment-780431068
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13398",
"triggerID" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"triggerType" : "PUSH"
}, {
"hash" : "bee61356ebeb5f99c96ce3193b3bd8205e61e44d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13409",
"triggerID" : "bee61356ebeb5f99c96ce3193b3bd8205e61e44d",
"triggerType" : "PUSH"
}, {
"hash" : "1b43a411d22af19f12f14a79c1094c8dc5cfeb42",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "1b43a411d22af19f12f14a79c1094c8dc5cfeb42",
"triggerType" : "PUSH"
}, {
"hash" : "e6aae42cf093d4f44b93b79a78bdc64538df8725",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13445",
"triggerID" : "e6aae42cf093d4f44b93b79a78bdc64538df8725",
"triggerType" : "PUSH"
}, {
"hash" : "cccca8191e6800961467c726695c8ef7575d157a",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13449",
"triggerID" : "cccca8191e6800961467c726695c8ef7575d157a",
"triggerType" : "PUSH"
}, {
"hash" : "44ee00afbd5ab0c81c53ed5787e7c0701b711e41",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "44ee00afbd5ab0c81c53ed5787e7c0701b711e41",
"triggerType" : "PUSH"
}, {
"hash" : "8298e7dcbbd9809429dd4c2d4148ff5e46bdb43b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13499",
"triggerID" : "8298e7dcbbd9809429dd4c2d4148ff5e46bdb43b",
"triggerType" : "PUSH"
}, {
"hash" : "d77a6c96e6616c2485ed787b7a1f0fedea97511b",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13599",
"triggerID" : "d77a6c96e6616c2485ed787b7a1f0fedea97511b",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1b43a411d22af19f12f14a79c1094c8dc5cfeb42 UNKNOWN
* 44ee00afbd5ab0c81c53ed5787e7c0701b711e41 UNKNOWN
* d77a6c96e6616c2485ed787b7a1f0fedea97511b Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13599)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] rmetzger commented on pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
rmetzger commented on pull request #14950:
URL: https://github.com/apache/flink/pull/14950#issuecomment-781246489
Rebased onto DeclScheduler merge on master
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] rmetzger commented on pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
rmetzger commented on pull request #14950:
URL: https://github.com/apache/flink/pull/14950#issuecomment-783594935
My understanding of your (valid) feedback was that it's a bad idea to expose private methods via the `ExecutionGraph` interface, that's why I tried moving all private methods (+ dependencies) into a separate interface.
The refactoring was mostly driven by feedback from the type checker, not so much from logical parts of the code.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14950:
URL: https://github.com/apache/flink/pull/14950#issuecomment-780431068
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13398",
"triggerID" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"triggerType" : "PUSH"
}, {
"hash" : "bee61356ebeb5f99c96ce3193b3bd8205e61e44d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13409",
"triggerID" : "bee61356ebeb5f99c96ce3193b3bd8205e61e44d",
"triggerType" : "PUSH"
}, {
"hash" : "1b43a411d22af19f12f14a79c1094c8dc5cfeb42",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "1b43a411d22af19f12f14a79c1094c8dc5cfeb42",
"triggerType" : "PUSH"
}, {
"hash" : "e6aae42cf093d4f44b93b79a78bdc64538df8725",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13445",
"triggerID" : "e6aae42cf093d4f44b93b79a78bdc64538df8725",
"triggerType" : "PUSH"
}, {
"hash" : "cccca8191e6800961467c726695c8ef7575d157a",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13449",
"triggerID" : "cccca8191e6800961467c726695c8ef7575d157a",
"triggerType" : "PUSH"
}, {
"hash" : "44ee00afbd5ab0c81c53ed5787e7c0701b711e41",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "44ee00afbd5ab0c81c53ed5787e7c0701b711e41",
"triggerType" : "PUSH"
}, {
"hash" : "8298e7dcbbd9809429dd4c2d4148ff5e46bdb43b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13499",
"triggerID" : "8298e7dcbbd9809429dd4c2d4148ff5e46bdb43b",
"triggerType" : "PUSH"
}, {
"hash" : "d77a6c96e6616c2485ed787b7a1f0fedea97511b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13599",
"triggerID" : "d77a6c96e6616c2485ed787b7a1f0fedea97511b",
"triggerType" : "PUSH"
}, {
"hash" : "c1274ae69962c108df47f9939b3454154724acbb",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14049",
"triggerID" : "c1274ae69962c108df47f9939b3454154724acbb",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1b43a411d22af19f12f14a79c1094c8dc5cfeb42 UNKNOWN
* 44ee00afbd5ab0c81c53ed5787e7c0701b711e41 UNKNOWN
* c1274ae69962c108df47f9939b3454154724acbb Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14049)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] zentol commented on a change in pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
zentol commented on a change in pull request #14950:
URL: https://github.com/apache/flink/pull/14950#discussion_r578976880
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/declarative/StateTrackingMockExecutionGraph.java
##########
@@ -0,0 +1,459 @@
+/*
+ * 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.declarative;
+
+import org.apache.flink.api.common.ArchivedExecutionConfig;
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.api.common.accumulators.Accumulator;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.JobException;
+import org.apache.flink.runtime.accumulators.AccumulatorSnapshot;
+import org.apache.flink.runtime.accumulators.StringifiedAccumulatorResult;
+import org.apache.flink.runtime.blob.BlobWriter;
+import org.apache.flink.runtime.blob.PermanentBlobKey;
+import org.apache.flink.runtime.checkpoint.CheckpointCoordinator;
+import org.apache.flink.runtime.checkpoint.CheckpointIDCounter;
+import org.apache.flink.runtime.checkpoint.CheckpointStatsSnapshot;
+import org.apache.flink.runtime.checkpoint.CheckpointStatsTracker;
+import org.apache.flink.runtime.checkpoint.CheckpointsCleaner;
+import org.apache.flink.runtime.checkpoint.CompletedCheckpointStore;
+import org.apache.flink.runtime.checkpoint.MasterTriggerRestoreHook;
+import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor;
+import org.apache.flink.runtime.execution.ExecutionState;
+import org.apache.flink.runtime.executiongraph.ErrorInfo;
+import org.apache.flink.runtime.executiongraph.Execution;
+import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
+import org.apache.flink.runtime.executiongraph.ExecutionDeploymentListener;
+import org.apache.flink.runtime.executiongraph.ExecutionGraph;
+import org.apache.flink.runtime.executiongraph.ExecutionJobVertex;
+import org.apache.flink.runtime.executiongraph.ExecutionVertex;
+import org.apache.flink.runtime.executiongraph.IntermediateResult;
+import org.apache.flink.runtime.executiongraph.JobInformation;
+import org.apache.flink.runtime.executiongraph.JobStatusListener;
+import org.apache.flink.runtime.executiongraph.TaskExecutionStateTransition;
+import org.apache.flink.runtime.executiongraph.failover.flip1.ResultPartitionAvailabilityChecker;
+import org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease.PartitionReleaseStrategy;
+import org.apache.flink.runtime.io.network.partition.JobMasterPartitionTracker;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
+import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
+import org.apache.flink.runtime.jobgraph.JobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+import org.apache.flink.runtime.jobgraph.ScheduleMode;
+import org.apache.flink.runtime.jobgraph.tasks.CheckpointCoordinatorConfiguration;
+import org.apache.flink.runtime.query.KvStateLocationRegistry;
+import org.apache.flink.runtime.scheduler.InternalFailuresListener;
+import org.apache.flink.runtime.scheduler.strategy.SchedulingTopology;
+import org.apache.flink.runtime.shuffle.ShuffleMaster;
+import org.apache.flink.runtime.state.CheckpointStorage;
+import org.apache.flink.runtime.state.StateBackend;
+import org.apache.flink.types.Either;
+import org.apache.flink.util.OptionalFailure;
+import org.apache.flink.util.SerializedValue;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ForkJoinPool;
+
+/**
+ * Mocked ExecutionGraph which (partially) tracks the job status, and provides some basic mocks to
+ * create an {@link org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph} from this
+ * ExecutionGraph.
+ */
+class StateTrackingMockExecutionGraph implements ExecutionGraph {
+ private JobStatus state = JobStatus.INITIALIZING;
+ private final CompletableFuture<JobStatus> terminationFuture = new CompletableFuture<>();
+ private final JobID jobId = new JobID();
+ private final ArchivedExecutionConfig archivedExecutionConfig = new ExecutionConfig().archive();
+
+ // ---- methods to control the mock
+
+ void completeTerminationFuture(JobStatus finalStatus) {
+ terminationFuture.complete(finalStatus);
+ state = finalStatus;
+ }
+
+ // ---- interface implementations
+
+ @Override
+ public boolean updateState(TaskExecutionStateTransition state) {
+ return true;
+ }
+
+ @Override
+ public JobStatus getState() {
+ return state;
+ }
+
+ @Override
+ public CompletableFuture<JobStatus> getTerminationFuture() {
+ return terminationFuture;
+ }
+
+ @Override
+ public Executor getFutureExecutor() {
+ return ForkJoinPool.commonPool();
+ }
+
+ @Override
+ public void registerExecution(Execution exec) {
+ // no-op
+ }
+
+ @Override
+ public void cancel() {
+ state = JobStatus.CANCELLING;
+ }
+
+ @Override
+ public void failJob(Throwable cause) {
+ state = JobStatus.FAILING;
+ }
+
+ @Override
+ public void suspend(Throwable suspensionCause) {
+ state = JobStatus.SUSPENDED;
+ }
+
+ @Override
+ public void transitionToRunning() {
+ state = JobStatus.RUNNING;
+ }
+
+ // --- interface implementations: methods for creating an archived execution graph
+
+ @Override
+ public int getTotalNumberOfVertices() {
+ return 0;
+ }
+
+ @Override
+ public Iterable<ExecutionJobVertex> getVerticesTopologically() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public Map<JobVertexID, ExecutionJobVertex> getAllVertices() {
+ return Collections.emptyMap();
+ }
+
+ @Override
+ public Map<String, SerializedValue<OptionalFailure<Object>>> getAccumulatorsSerialized() {
+ return Collections.emptyMap();
+ }
+
+ @Override
+ public long getStatusTimestamp(JobStatus status) {
+ return 0L;
+ }
+
+ @Override
+ public String getJsonPlan() {
+ return "";
+ }
+
+ @Override
+ public JobID getJobID() {
+ return jobId;
+ }
+
+ @Override
+ public String getJobName() {
+ return "testJob";
+ }
+
+ @Nullable
+ @Override
+ public ErrorInfo getFailureInfo() {
+ return null;
+ }
+
+ @Nullable
+ @Override
+ public ArchivedExecutionConfig getArchivedExecutionConfig() {
+ return archivedExecutionConfig;
+ }
+
+ @Override
+ public boolean isStoppable() {
+ return false;
+ }
+
+ @Nullable
+ @Override
+ public CheckpointCoordinatorConfiguration getCheckpointCoordinatorConfiguration() {
+ return null;
+ }
+
+ @Nullable
+ @Override
+ public CheckpointStatsSnapshot getCheckpointStatsSnapshot() {
+ return null;
+ }
+
+ @Override
+ public Optional<String> getStateBackendName() {
+ return Optional.empty();
+ }
+
+ @Override
+ public Optional<String> getCheckpointStorageName() {
+ return Optional.empty();
+ }
+
+ @Override
+ public StringifiedAccumulatorResult[] getAccumulatorResultsStringified() {
+ return new StringifiedAccumulatorResult[0];
+ }
+
+ // -- remaining interface implementations: all unsupported
+
+ @Override
+ public boolean isArchived() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void start(@Nonnull ComponentMainThreadExecutor jobMasterMainThreadExecutor) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public int getNumberOfExecutionJobVertices() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public SchedulingTopology getSchedulingTopology() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public ScheduleMode getScheduleMode() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Nonnull
+ @Override
+ public ComponentMainThreadExecutor getJobMasterMainThreadExecutor() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void enableCheckpointing(
+ CheckpointCoordinatorConfiguration chkConfig,
+ List<MasterTriggerRestoreHook<?>> masterHooks,
+ CheckpointIDCounter checkpointIDCounter,
+ CompletedCheckpointStore checkpointStore,
+ StateBackend checkpointStateBackend,
+ CheckpointStorage checkpointStorage,
+ CheckpointStatsTracker statsTracker,
+ CheckpointsCleaner checkpointsCleaner) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Nullable
+ @Override
+ public CheckpointCoordinator getCheckpointCoordinator() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public KvStateLocationRegistry getKvStateLocationRegistry() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void setJsonPlan(String jsonPlan) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Either<SerializedValue<JobInformation>, PermanentBlobKey> getJobInformationOrBlobKey() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Configuration getJobConfiguration() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public ClassLoader getUserClassLoader() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Throwable getFailureCause() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Iterable<ExecutionVertex> getAllExecutionVertices() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public ExecutionJobVertex getJobVertex(JobVertexID id) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public long getNumberOfRestarts() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Map<IntermediateDataSetID, IntermediateResult> getAllIntermediateResults() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public BlobWriter getBlobWriter() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Map<String, OptionalFailure<Accumulator<?, ?>>> aggregateUserAccumulators() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void setInternalTaskFailuresListener(
+ InternalFailuresListener internalTaskFailuresListener) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void attachJobGraph(List<JobVertex> topologiallySorted) throws JobException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void failGlobal(Throwable t) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public JobStatus waitUntilTerminal() throws InterruptedException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean transitionState(JobStatus current, JobStatus newState) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void incrementRestarts() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void initFailureCause(Throwable t) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void vertexFinished() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void vertexUnFinished() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void notifyPartitionDataAvailable(ResultPartitionID partitionId) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Map<ExecutionAttemptID, Execution> getRegisteredExecutions() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateAccumulators(AccumulatorSnapshot accumulatorSnapshot) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void registerJobStatusListener(JobStatusListener listener) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void assertRunningInJobMasterMainThread() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void notifySchedulerNgAboutInternalTaskFailure(
+ ExecutionAttemptID attemptId,
+ Throwable t,
+ boolean cancelTask,
+ boolean releasePartitions) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public ShuffleMaster<?> getShuffleMaster() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public JobMasterPartitionTracker getPartitionTracker() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public ResultPartitionAvailabilityChecker getResultPartitionAvailabilityChecker() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public PartitionReleaseStrategy getPartitionReleaseStrategy() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public ExecutionDeploymentListener getExecutionDeploymentListener() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void deregisterExecution(Execution exec) {
+ throw new UnsupportedOperationException();
Review comment:
its a bit odd that registering an execution is a no-op but deregistering fails
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] zentol commented on pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
zentol commented on pull request #14950:
URL: https://github.com/apache/flink/pull/14950#issuecomment-782884068
Could you expand on why you went with introducing an interface between the vertices and EG, instead of the Executing state and vertices?
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] rmetzger commented on a change in pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
rmetzger commented on a change in pull request #14950:
URL: https://github.com/apache/flink/pull/14950#discussion_r586225417
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphDeploymentWithBlobServerTest.java
##########
@@ -87,7 +87,7 @@ public void shutdownBlobServer() throws IOException {
}
@Override
- protected void checkJobOffloaded(ExecutionGraph eg) throws Exception {
+ protected void checkJobOffloaded(DefaultExecutionGraph eg) throws Exception {
Review comment:
Let me know if you believe it should be part of the interface. To me it seems to be fine to hide it. Not sure if one test usage justifies exposing it.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #14950:
URL: https://github.com/apache/flink/pull/14950#issuecomment-780426957
Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.
## Automated Checks
Last check on commit e44c9cdb2e60872179635616e8ffb184bb0337b9 (Wed Feb 17 09:33:40 UTC 2021)
**Warnings:**
* No documentation files were touched! Remember to keep the Flink docs up to date!
<sub>Mention the bot in a comment to re-run the automated checks.</sub>
## Review Progress
* ❓ 1. The [description] looks good.
* ❓ 2. There is [consensus] that the contribution should go into to Flink.
* ❓ 3. Needs [attention] from.
* ❓ 4. The change fits into the overall [architecture].
* ❓ 5. Overall code [quality] is good.
Please see the [Pull Request Review Guide](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process.<details>
The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required <summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot approve description` to approve one or more aspects (aspects: `description`, `consensus`, `architecture` and `quality`)
- `@flinkbot approve all` to approve all aspects
- `@flinkbot approve-until architecture` to approve everything until `architecture`
- `@flinkbot attention @username1 [@username2 ..]` to require somebody's attention
- `@flinkbot disapprove architecture` to remove an approval you gave earlier
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14950:
URL: https://github.com/apache/flink/pull/14950#issuecomment-780431068
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13398",
"triggerID" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"triggerType" : "PUSH"
}, {
"hash" : "bee61356ebeb5f99c96ce3193b3bd8205e61e44d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13409",
"triggerID" : "bee61356ebeb5f99c96ce3193b3bd8205e61e44d",
"triggerType" : "PUSH"
}, {
"hash" : "1b43a411d22af19f12f14a79c1094c8dc5cfeb42",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "1b43a411d22af19f12f14a79c1094c8dc5cfeb42",
"triggerType" : "PUSH"
}, {
"hash" : "e6aae42cf093d4f44b93b79a78bdc64538df8725",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13445",
"triggerID" : "e6aae42cf093d4f44b93b79a78bdc64538df8725",
"triggerType" : "PUSH"
}, {
"hash" : "cccca8191e6800961467c726695c8ef7575d157a",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13449",
"triggerID" : "cccca8191e6800961467c726695c8ef7575d157a",
"triggerType" : "PUSH"
}, {
"hash" : "44ee00afbd5ab0c81c53ed5787e7c0701b711e41",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "44ee00afbd5ab0c81c53ed5787e7c0701b711e41",
"triggerType" : "PUSH"
}, {
"hash" : "8298e7dcbbd9809429dd4c2d4148ff5e46bdb43b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13499",
"triggerID" : "8298e7dcbbd9809429dd4c2d4148ff5e46bdb43b",
"triggerType" : "PUSH"
}, {
"hash" : "d77a6c96e6616c2485ed787b7a1f0fedea97511b",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13599",
"triggerID" : "d77a6c96e6616c2485ed787b7a1f0fedea97511b",
"triggerType" : "PUSH"
}, {
"hash" : "c1274ae69962c108df47f9939b3454154724acbb",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "c1274ae69962c108df47f9939b3454154724acbb",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1b43a411d22af19f12f14a79c1094c8dc5cfeb42 UNKNOWN
* 44ee00afbd5ab0c81c53ed5787e7c0701b711e41 UNKNOWN
* d77a6c96e6616c2485ed787b7a1f0fedea97511b Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13599)
* c1274ae69962c108df47f9939b3454154724acbb UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] rmetzger commented on a change in pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
rmetzger commented on a change in pull request #14950:
URL: https://github.com/apache/flink/pull/14950#discussion_r578985730
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/declarative/StateTrackingMockExecutionGraph.java
##########
@@ -0,0 +1,459 @@
+/*
+ * 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.declarative;
+
+import org.apache.flink.api.common.ArchivedExecutionConfig;
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.api.common.accumulators.Accumulator;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.JobException;
+import org.apache.flink.runtime.accumulators.AccumulatorSnapshot;
+import org.apache.flink.runtime.accumulators.StringifiedAccumulatorResult;
+import org.apache.flink.runtime.blob.BlobWriter;
+import org.apache.flink.runtime.blob.PermanentBlobKey;
+import org.apache.flink.runtime.checkpoint.CheckpointCoordinator;
+import org.apache.flink.runtime.checkpoint.CheckpointIDCounter;
+import org.apache.flink.runtime.checkpoint.CheckpointStatsSnapshot;
+import org.apache.flink.runtime.checkpoint.CheckpointStatsTracker;
+import org.apache.flink.runtime.checkpoint.CheckpointsCleaner;
+import org.apache.flink.runtime.checkpoint.CompletedCheckpointStore;
+import org.apache.flink.runtime.checkpoint.MasterTriggerRestoreHook;
+import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor;
+import org.apache.flink.runtime.execution.ExecutionState;
+import org.apache.flink.runtime.executiongraph.ErrorInfo;
+import org.apache.flink.runtime.executiongraph.Execution;
+import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
+import org.apache.flink.runtime.executiongraph.ExecutionDeploymentListener;
+import org.apache.flink.runtime.executiongraph.ExecutionGraph;
+import org.apache.flink.runtime.executiongraph.ExecutionJobVertex;
+import org.apache.flink.runtime.executiongraph.ExecutionVertex;
+import org.apache.flink.runtime.executiongraph.IntermediateResult;
+import org.apache.flink.runtime.executiongraph.JobInformation;
+import org.apache.flink.runtime.executiongraph.JobStatusListener;
+import org.apache.flink.runtime.executiongraph.TaskExecutionStateTransition;
+import org.apache.flink.runtime.executiongraph.failover.flip1.ResultPartitionAvailabilityChecker;
+import org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease.PartitionReleaseStrategy;
+import org.apache.flink.runtime.io.network.partition.JobMasterPartitionTracker;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
+import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
+import org.apache.flink.runtime.jobgraph.JobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+import org.apache.flink.runtime.jobgraph.ScheduleMode;
+import org.apache.flink.runtime.jobgraph.tasks.CheckpointCoordinatorConfiguration;
+import org.apache.flink.runtime.query.KvStateLocationRegistry;
+import org.apache.flink.runtime.scheduler.InternalFailuresListener;
+import org.apache.flink.runtime.scheduler.strategy.SchedulingTopology;
+import org.apache.flink.runtime.shuffle.ShuffleMaster;
+import org.apache.flink.runtime.state.CheckpointStorage;
+import org.apache.flink.runtime.state.StateBackend;
+import org.apache.flink.types.Either;
+import org.apache.flink.util.OptionalFailure;
+import org.apache.flink.util.SerializedValue;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ForkJoinPool;
+
+/**
+ * Mocked ExecutionGraph which (partially) tracks the job status, and provides some basic mocks to
+ * create an {@link org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph} from this
+ * ExecutionGraph.
+ */
+class StateTrackingMockExecutionGraph implements ExecutionGraph {
+ private JobStatus state = JobStatus.INITIALIZING;
+ private final CompletableFuture<JobStatus> terminationFuture = new CompletableFuture<>();
+ private final JobID jobId = new JobID();
+ private final ArchivedExecutionConfig archivedExecutionConfig = new ExecutionConfig().archive();
+
+ // ---- methods to control the mock
+
+ void completeTerminationFuture(JobStatus finalStatus) {
+ terminationFuture.complete(finalStatus);
+ state = finalStatus;
+ }
+
+ // ---- interface implementations
+
+ @Override
+ public boolean updateState(TaskExecutionStateTransition state) {
+ return true;
+ }
+
+ @Override
+ public JobStatus getState() {
+ return state;
+ }
+
+ @Override
+ public CompletableFuture<JobStatus> getTerminationFuture() {
+ return terminationFuture;
+ }
+
+ @Override
+ public Executor getFutureExecutor() {
+ return ForkJoinPool.commonPool();
+ }
+
+ @Override
+ public void registerExecution(Execution exec) {
+ // no-op
+ }
+
+ @Override
+ public void cancel() {
+ state = JobStatus.CANCELLING;
+ }
+
+ @Override
+ public void failJob(Throwable cause) {
+ state = JobStatus.FAILING;
+ }
+
+ @Override
+ public void suspend(Throwable suspensionCause) {
+ state = JobStatus.SUSPENDED;
+ }
+
+ @Override
+ public void transitionToRunning() {
+ state = JobStatus.RUNNING;
+ }
+
+ // --- interface implementations: methods for creating an archived execution graph
+
+ @Override
+ public int getTotalNumberOfVertices() {
+ return 0;
+ }
+
+ @Override
+ public Iterable<ExecutionJobVertex> getVerticesTopologically() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public Map<JobVertexID, ExecutionJobVertex> getAllVertices() {
+ return Collections.emptyMap();
+ }
+
+ @Override
+ public Map<String, SerializedValue<OptionalFailure<Object>>> getAccumulatorsSerialized() {
+ return Collections.emptyMap();
+ }
+
+ @Override
+ public long getStatusTimestamp(JobStatus status) {
+ return 0L;
+ }
+
+ @Override
+ public String getJsonPlan() {
+ return "";
+ }
+
+ @Override
+ public JobID getJobID() {
+ return jobId;
+ }
+
+ @Override
+ public String getJobName() {
+ return "testJob";
+ }
+
+ @Nullable
+ @Override
+ public ErrorInfo getFailureInfo() {
+ return null;
+ }
+
+ @Nullable
+ @Override
+ public ArchivedExecutionConfig getArchivedExecutionConfig() {
+ return archivedExecutionConfig;
+ }
+
+ @Override
+ public boolean isStoppable() {
+ return false;
+ }
+
+ @Nullable
+ @Override
+ public CheckpointCoordinatorConfiguration getCheckpointCoordinatorConfiguration() {
+ return null;
+ }
+
+ @Nullable
+ @Override
+ public CheckpointStatsSnapshot getCheckpointStatsSnapshot() {
+ return null;
+ }
+
+ @Override
+ public Optional<String> getStateBackendName() {
+ return Optional.empty();
+ }
+
+ @Override
+ public Optional<String> getCheckpointStorageName() {
+ return Optional.empty();
+ }
+
+ @Override
+ public StringifiedAccumulatorResult[] getAccumulatorResultsStringified() {
+ return new StringifiedAccumulatorResult[0];
+ }
+
+ // -- remaining interface implementations: all unsupported
+
+ @Override
+ public boolean isArchived() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void start(@Nonnull ComponentMainThreadExecutor jobMasterMainThreadExecutor) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public int getNumberOfExecutionJobVertices() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public SchedulingTopology getSchedulingTopology() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public ScheduleMode getScheduleMode() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Nonnull
+ @Override
+ public ComponentMainThreadExecutor getJobMasterMainThreadExecutor() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void enableCheckpointing(
+ CheckpointCoordinatorConfiguration chkConfig,
+ List<MasterTriggerRestoreHook<?>> masterHooks,
+ CheckpointIDCounter checkpointIDCounter,
+ CompletedCheckpointStore checkpointStore,
+ StateBackend checkpointStateBackend,
+ CheckpointStorage checkpointStorage,
+ CheckpointStatsTracker statsTracker,
+ CheckpointsCleaner checkpointsCleaner) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Nullable
+ @Override
+ public CheckpointCoordinator getCheckpointCoordinator() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public KvStateLocationRegistry getKvStateLocationRegistry() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void setJsonPlan(String jsonPlan) {
+ throw new UnsupportedOperationException();
Review comment:
Yes, but why should I spend time on something I don't need?
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14950:
URL: https://github.com/apache/flink/pull/14950#issuecomment-780431068
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13398",
"triggerID" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"triggerType" : "PUSH"
}, {
"hash" : "bee61356ebeb5f99c96ce3193b3bd8205e61e44d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13409",
"triggerID" : "bee61356ebeb5f99c96ce3193b3bd8205e61e44d",
"triggerType" : "PUSH"
}, {
"hash" : "1b43a411d22af19f12f14a79c1094c8dc5cfeb42",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "1b43a411d22af19f12f14a79c1094c8dc5cfeb42",
"triggerType" : "PUSH"
}, {
"hash" : "e6aae42cf093d4f44b93b79a78bdc64538df8725",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13445",
"triggerID" : "e6aae42cf093d4f44b93b79a78bdc64538df8725",
"triggerType" : "PUSH"
}, {
"hash" : "cccca8191e6800961467c726695c8ef7575d157a",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13449",
"triggerID" : "cccca8191e6800961467c726695c8ef7575d157a",
"triggerType" : "PUSH"
}, {
"hash" : "44ee00afbd5ab0c81c53ed5787e7c0701b711e41",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "44ee00afbd5ab0c81c53ed5787e7c0701b711e41",
"triggerType" : "PUSH"
}, {
"hash" : "8298e7dcbbd9809429dd4c2d4148ff5e46bdb43b",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "8298e7dcbbd9809429dd4c2d4148ff5e46bdb43b",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1b43a411d22af19f12f14a79c1094c8dc5cfeb42 UNKNOWN
* cccca8191e6800961467c726695c8ef7575d157a Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13449)
* 44ee00afbd5ab0c81c53ed5787e7c0701b711e41 UNKNOWN
* 8298e7dcbbd9809429dd4c2d4148ff5e46bdb43b UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14950:
URL: https://github.com/apache/flink/pull/14950#issuecomment-780431068
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13398",
"triggerID" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"triggerType" : "PUSH"
}, {
"hash" : "bee61356ebeb5f99c96ce3193b3bd8205e61e44d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13409",
"triggerID" : "bee61356ebeb5f99c96ce3193b3bd8205e61e44d",
"triggerType" : "PUSH"
}, {
"hash" : "1b43a411d22af19f12f14a79c1094c8dc5cfeb42",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "1b43a411d22af19f12f14a79c1094c8dc5cfeb42",
"triggerType" : "PUSH"
}, {
"hash" : "e6aae42cf093d4f44b93b79a78bdc64538df8725",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13445",
"triggerID" : "e6aae42cf093d4f44b93b79a78bdc64538df8725",
"triggerType" : "PUSH"
}, {
"hash" : "cccca8191e6800961467c726695c8ef7575d157a",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13449",
"triggerID" : "cccca8191e6800961467c726695c8ef7575d157a",
"triggerType" : "PUSH"
}, {
"hash" : "44ee00afbd5ab0c81c53ed5787e7c0701b711e41",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "44ee00afbd5ab0c81c53ed5787e7c0701b711e41",
"triggerType" : "PUSH"
}, {
"hash" : "8298e7dcbbd9809429dd4c2d4148ff5e46bdb43b",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13499",
"triggerID" : "8298e7dcbbd9809429dd4c2d4148ff5e46bdb43b",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1b43a411d22af19f12f14a79c1094c8dc5cfeb42 UNKNOWN
* cccca8191e6800961467c726695c8ef7575d157a Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13449)
* 44ee00afbd5ab0c81c53ed5787e7c0701b711e41 UNKNOWN
* 8298e7dcbbd9809429dd4c2d4148ff5e46bdb43b Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13499)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] rmetzger commented on a change in pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
rmetzger commented on a change in pull request #14950:
URL: https://github.com/apache/flink/pull/14950#discussion_r578994044
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
##########
@@ -1021,203 +220,20 @@ public ArchivedExecutionConfig getArchivedExecutionConfig() {
*
* @return Termination future of this {@link ExecutionGraph}.
*/
- public CompletableFuture<JobStatus> getTerminationFuture() {
- return terminationFuture;
- }
+ CompletableFuture<JobStatus> getTerminationFuture();
@VisibleForTesting
- public JobStatus waitUntilTerminal() throws InterruptedException {
- try {
- return terminationFuture.get();
- } catch (ExecutionException e) {
- // this should never happen
- // it would be a bug, so we don't expect this to be handled and throw
- // an unchecked exception here
- throw new RuntimeException(e);
- }
- }
-
- // ------------------------------------------------------------------------
- // State Transitions
- // ------------------------------------------------------------------------
-
- public boolean transitionState(JobStatus current, JobStatus newState) {
- return transitionState(current, newState, null);
- }
-
- private void transitionState(JobStatus newState, Throwable error) {
- transitionState(state, newState, error);
- }
+ JobStatus waitUntilTerminal() throws InterruptedException;
- private boolean transitionState(JobStatus current, JobStatus newState, Throwable error) {
- assertRunningInJobMasterMainThread();
- // consistency check
- if (current.isTerminalState()) {
- String message = "Job is trying to leave terminal state " + current;
- LOG.error(message);
- throw new IllegalStateException(message);
- }
+ boolean transitionState(JobStatus current, JobStatus newState);
- // now do the actual state transition
- if (state == current) {
- state = newState;
- LOG.info(
- "Job {} ({}) switched from state {} to {}.",
- getJobName(),
- getJobID(),
- current,
- newState,
- error);
+ void incrementRestarts();
- stateTimestamps[newState.ordinal()] = System.currentTimeMillis();
- notifyJobStatusChange(newState, error);
- return true;
- } else {
- return false;
- }
- }
+ void initFailureCause(Throwable t);
- public void incrementRestarts() {
- numberOfRestartsCounter.inc();
- }
+ void vertexFinished();
Review comment:
Yes, I considered this as well. But the problem is that this introduces a dependency on DefaultExecutionGraph again.
For example here: https://github.com/apache/flink/blob/master/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/declarative/ExecutingTest.java#L672 , we would require a DefaultExecutionGraph in a test mock, voiding the benefits of this effort.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14950:
URL: https://github.com/apache/flink/pull/14950#issuecomment-780431068
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13398",
"triggerID" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"triggerType" : "PUSH"
}, {
"hash" : "bee61356ebeb5f99c96ce3193b3bd8205e61e44d",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13409",
"triggerID" : "bee61356ebeb5f99c96ce3193b3bd8205e61e44d",
"triggerType" : "PUSH"
}, {
"hash" : "1b43a411d22af19f12f14a79c1094c8dc5cfeb42",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "1b43a411d22af19f12f14a79c1094c8dc5cfeb42",
"triggerType" : "PUSH"
}, {
"hash" : "e6aae42cf093d4f44b93b79a78bdc64538df8725",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "e6aae42cf093d4f44b93b79a78bdc64538df8725",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* bee61356ebeb5f99c96ce3193b3bd8205e61e44d Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13409)
* 1b43a411d22af19f12f14a79c1094c8dc5cfeb42 UNKNOWN
* e6aae42cf093d4f44b93b79a78bdc64538df8725 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14950:
URL: https://github.com/apache/flink/pull/14950#issuecomment-780431068
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13398",
"triggerID" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"triggerType" : "PUSH"
}, {
"hash" : "bee61356ebeb5f99c96ce3193b3bd8205e61e44d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13409",
"triggerID" : "bee61356ebeb5f99c96ce3193b3bd8205e61e44d",
"triggerType" : "PUSH"
}, {
"hash" : "1b43a411d22af19f12f14a79c1094c8dc5cfeb42",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "1b43a411d22af19f12f14a79c1094c8dc5cfeb42",
"triggerType" : "PUSH"
}, {
"hash" : "e6aae42cf093d4f44b93b79a78bdc64538df8725",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13445",
"triggerID" : "e6aae42cf093d4f44b93b79a78bdc64538df8725",
"triggerType" : "PUSH"
}, {
"hash" : "cccca8191e6800961467c726695c8ef7575d157a",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13449",
"triggerID" : "cccca8191e6800961467c726695c8ef7575d157a",
"triggerType" : "PUSH"
}, {
"hash" : "44ee00afbd5ab0c81c53ed5787e7c0701b711e41",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "44ee00afbd5ab0c81c53ed5787e7c0701b711e41",
"triggerType" : "PUSH"
}, {
"hash" : "8298e7dcbbd9809429dd4c2d4148ff5e46bdb43b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13499",
"triggerID" : "8298e7dcbbd9809429dd4c2d4148ff5e46bdb43b",
"triggerType" : "PUSH"
}, {
"hash" : "d77a6c96e6616c2485ed787b7a1f0fedea97511b",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13599",
"triggerID" : "d77a6c96e6616c2485ed787b7a1f0fedea97511b",
"triggerType" : "PUSH"
}, {
"hash" : "c1274ae69962c108df47f9939b3454154724acbb",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14049",
"triggerID" : "c1274ae69962c108df47f9939b3454154724acbb",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1b43a411d22af19f12f14a79c1094c8dc5cfeb42 UNKNOWN
* 44ee00afbd5ab0c81c53ed5787e7c0701b711e41 UNKNOWN
* d77a6c96e6616c2485ed787b7a1f0fedea97511b Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13599)
* c1274ae69962c108df47f9939b3454154724acbb Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14049)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] zentol commented on a change in pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
zentol commented on a change in pull request #14950:
URL: https://github.com/apache/flink/pull/14950#discussion_r578978899
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
##########
@@ -1021,203 +220,20 @@ public ArchivedExecutionConfig getArchivedExecutionConfig() {
*
* @return Termination future of this {@link ExecutionGraph}.
*/
- public CompletableFuture<JobStatus> getTerminationFuture() {
- return terminationFuture;
- }
+ CompletableFuture<JobStatus> getTerminationFuture();
@VisibleForTesting
- public JobStatus waitUntilTerminal() throws InterruptedException {
- try {
- return terminationFuture.get();
- } catch (ExecutionException e) {
- // this should never happen
- // it would be a bug, so we don't expect this to be handled and throw
- // an unchecked exception here
- throw new RuntimeException(e);
- }
- }
-
- // ------------------------------------------------------------------------
- // State Transitions
- // ------------------------------------------------------------------------
-
- public boolean transitionState(JobStatus current, JobStatus newState) {
- return transitionState(current, newState, null);
- }
-
- private void transitionState(JobStatus newState, Throwable error) {
- transitionState(state, newState, error);
- }
+ JobStatus waitUntilTerminal() throws InterruptedException;
- private boolean transitionState(JobStatus current, JobStatus newState, Throwable error) {
- assertRunningInJobMasterMainThread();
- // consistency check
- if (current.isTerminalState()) {
- String message = "Job is trying to leave terminal state " + current;
- LOG.error(message);
- throw new IllegalStateException(message);
- }
+ boolean transitionState(JobStatus current, JobStatus newState);
- // now do the actual state transition
- if (state == current) {
- state = newState;
- LOG.info(
- "Job {} ({}) switched from state {} to {}.",
- getJobName(),
- getJobID(),
- current,
- newState,
- error);
+ void incrementRestarts();
- stateTimestamps[newState.ordinal()] = System.currentTimeMillis();
- notifyJobStatusChange(newState, error);
- return true;
- } else {
- return false;
- }
- }
+ void initFailureCause(Throwable t);
- public void incrementRestarts() {
- numberOfRestartsCounter.inc();
- }
+ void vertexFinished();
Review comment:
If we made sure that the Execution* classes all work against the DefaultExecutionGraph then we should be able to remove this and other package-private methods.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14950:
URL: https://github.com/apache/flink/pull/14950#issuecomment-780431068
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13398",
"triggerID" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"triggerType" : "PUSH"
}, {
"hash" : "bee61356ebeb5f99c96ce3193b3bd8205e61e44d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13409",
"triggerID" : "bee61356ebeb5f99c96ce3193b3bd8205e61e44d",
"triggerType" : "PUSH"
}, {
"hash" : "1b43a411d22af19f12f14a79c1094c8dc5cfeb42",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "1b43a411d22af19f12f14a79c1094c8dc5cfeb42",
"triggerType" : "PUSH"
}, {
"hash" : "e6aae42cf093d4f44b93b79a78bdc64538df8725",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13445",
"triggerID" : "e6aae42cf093d4f44b93b79a78bdc64538df8725",
"triggerType" : "PUSH"
}, {
"hash" : "cccca8191e6800961467c726695c8ef7575d157a",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13449",
"triggerID" : "cccca8191e6800961467c726695c8ef7575d157a",
"triggerType" : "PUSH"
}, {
"hash" : "44ee00afbd5ab0c81c53ed5787e7c0701b711e41",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "44ee00afbd5ab0c81c53ed5787e7c0701b711e41",
"triggerType" : "PUSH"
}, {
"hash" : "8298e7dcbbd9809429dd4c2d4148ff5e46bdb43b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13499",
"triggerID" : "8298e7dcbbd9809429dd4c2d4148ff5e46bdb43b",
"triggerType" : "PUSH"
}, {
"hash" : "d77a6c96e6616c2485ed787b7a1f0fedea97511b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13599",
"triggerID" : "d77a6c96e6616c2485ed787b7a1f0fedea97511b",
"triggerType" : "PUSH"
}, {
"hash" : "c1274ae69962c108df47f9939b3454154724acbb",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14049",
"triggerID" : "c1274ae69962c108df47f9939b3454154724acbb",
"triggerType" : "PUSH"
}, {
"hash" : "03ac0474397f9055837dc5727d634ac42ee18ce7",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14071",
"triggerID" : "03ac0474397f9055837dc5727d634ac42ee18ce7",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1b43a411d22af19f12f14a79c1094c8dc5cfeb42 UNKNOWN
* 44ee00afbd5ab0c81c53ed5787e7c0701b711e41 UNKNOWN
* 03ac0474397f9055837dc5727d634ac42ee18ce7 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14071)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] zentol commented on a change in pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
zentol commented on a change in pull request #14950:
URL: https://github.com/apache/flink/pull/14950#discussion_r578996744
##########
File path: flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/declarative/StateTrackingMockExecutionGraph.java
##########
@@ -0,0 +1,459 @@
+/*
+ * 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.declarative;
+
+import org.apache.flink.api.common.ArchivedExecutionConfig;
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.api.common.accumulators.Accumulator;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.JobException;
+import org.apache.flink.runtime.accumulators.AccumulatorSnapshot;
+import org.apache.flink.runtime.accumulators.StringifiedAccumulatorResult;
+import org.apache.flink.runtime.blob.BlobWriter;
+import org.apache.flink.runtime.blob.PermanentBlobKey;
+import org.apache.flink.runtime.checkpoint.CheckpointCoordinator;
+import org.apache.flink.runtime.checkpoint.CheckpointIDCounter;
+import org.apache.flink.runtime.checkpoint.CheckpointStatsSnapshot;
+import org.apache.flink.runtime.checkpoint.CheckpointStatsTracker;
+import org.apache.flink.runtime.checkpoint.CheckpointsCleaner;
+import org.apache.flink.runtime.checkpoint.CompletedCheckpointStore;
+import org.apache.flink.runtime.checkpoint.MasterTriggerRestoreHook;
+import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor;
+import org.apache.flink.runtime.execution.ExecutionState;
+import org.apache.flink.runtime.executiongraph.ErrorInfo;
+import org.apache.flink.runtime.executiongraph.Execution;
+import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
+import org.apache.flink.runtime.executiongraph.ExecutionDeploymentListener;
+import org.apache.flink.runtime.executiongraph.ExecutionGraph;
+import org.apache.flink.runtime.executiongraph.ExecutionJobVertex;
+import org.apache.flink.runtime.executiongraph.ExecutionVertex;
+import org.apache.flink.runtime.executiongraph.IntermediateResult;
+import org.apache.flink.runtime.executiongraph.JobInformation;
+import org.apache.flink.runtime.executiongraph.JobStatusListener;
+import org.apache.flink.runtime.executiongraph.TaskExecutionStateTransition;
+import org.apache.flink.runtime.executiongraph.failover.flip1.ResultPartitionAvailabilityChecker;
+import org.apache.flink.runtime.executiongraph.failover.flip1.partitionrelease.PartitionReleaseStrategy;
+import org.apache.flink.runtime.io.network.partition.JobMasterPartitionTracker;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
+import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
+import org.apache.flink.runtime.jobgraph.JobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+import org.apache.flink.runtime.jobgraph.ScheduleMode;
+import org.apache.flink.runtime.jobgraph.tasks.CheckpointCoordinatorConfiguration;
+import org.apache.flink.runtime.query.KvStateLocationRegistry;
+import org.apache.flink.runtime.scheduler.InternalFailuresListener;
+import org.apache.flink.runtime.scheduler.strategy.SchedulingTopology;
+import org.apache.flink.runtime.shuffle.ShuffleMaster;
+import org.apache.flink.runtime.state.CheckpointStorage;
+import org.apache.flink.runtime.state.StateBackend;
+import org.apache.flink.types.Either;
+import org.apache.flink.util.OptionalFailure;
+import org.apache.flink.util.SerializedValue;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ForkJoinPool;
+
+/**
+ * Mocked ExecutionGraph which (partially) tracks the job status, and provides some basic mocks to
+ * create an {@link org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph} from this
+ * ExecutionGraph.
+ */
+class StateTrackingMockExecutionGraph implements ExecutionGraph {
+ private JobStatus state = JobStatus.INITIALIZING;
+ private final CompletableFuture<JobStatus> terminationFuture = new CompletableFuture<>();
+ private final JobID jobId = new JobID();
+ private final ArchivedExecutionConfig archivedExecutionConfig = new ExecutionConfig().archive();
+
+ // ---- methods to control the mock
+
+ void completeTerminationFuture(JobStatus finalStatus) {
+ terminationFuture.complete(finalStatus);
+ state = finalStatus;
+ }
+
+ // ---- interface implementations
+
+ @Override
+ public boolean updateState(TaskExecutionStateTransition state) {
+ return true;
+ }
+
+ @Override
+ public JobStatus getState() {
+ return state;
+ }
+
+ @Override
+ public CompletableFuture<JobStatus> getTerminationFuture() {
+ return terminationFuture;
+ }
+
+ @Override
+ public Executor getFutureExecutor() {
+ return ForkJoinPool.commonPool();
+ }
+
+ @Override
+ public void registerExecution(Execution exec) {
+ // no-op
+ }
+
+ @Override
+ public void cancel() {
+ state = JobStatus.CANCELLING;
+ }
+
+ @Override
+ public void failJob(Throwable cause) {
+ state = JobStatus.FAILING;
+ }
+
+ @Override
+ public void suspend(Throwable suspensionCause) {
+ state = JobStatus.SUSPENDED;
+ }
+
+ @Override
+ public void transitionToRunning() {
+ state = JobStatus.RUNNING;
+ }
+
+ // --- interface implementations: methods for creating an archived execution graph
+
+ @Override
+ public int getTotalNumberOfVertices() {
+ return 0;
+ }
+
+ @Override
+ public Iterable<ExecutionJobVertex> getVerticesTopologically() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public Map<JobVertexID, ExecutionJobVertex> getAllVertices() {
+ return Collections.emptyMap();
+ }
+
+ @Override
+ public Map<String, SerializedValue<OptionalFailure<Object>>> getAccumulatorsSerialized() {
+ return Collections.emptyMap();
+ }
+
+ @Override
+ public long getStatusTimestamp(JobStatus status) {
+ return 0L;
+ }
+
+ @Override
+ public String getJsonPlan() {
+ return "";
+ }
+
+ @Override
+ public JobID getJobID() {
+ return jobId;
+ }
+
+ @Override
+ public String getJobName() {
+ return "testJob";
+ }
+
+ @Nullable
+ @Override
+ public ErrorInfo getFailureInfo() {
+ return null;
+ }
+
+ @Nullable
+ @Override
+ public ArchivedExecutionConfig getArchivedExecutionConfig() {
+ return archivedExecutionConfig;
+ }
+
+ @Override
+ public boolean isStoppable() {
+ return false;
+ }
+
+ @Nullable
+ @Override
+ public CheckpointCoordinatorConfiguration getCheckpointCoordinatorConfiguration() {
+ return null;
+ }
+
+ @Nullable
+ @Override
+ public CheckpointStatsSnapshot getCheckpointStatsSnapshot() {
+ return null;
+ }
+
+ @Override
+ public Optional<String> getStateBackendName() {
+ return Optional.empty();
+ }
+
+ @Override
+ public Optional<String> getCheckpointStorageName() {
+ return Optional.empty();
+ }
+
+ @Override
+ public StringifiedAccumulatorResult[] getAccumulatorResultsStringified() {
+ return new StringifiedAccumulatorResult[0];
+ }
+
+ // -- remaining interface implementations: all unsupported
+
+ @Override
+ public boolean isArchived() {
+ throw new UnsupportedOperationException();
Review comment:
I mean to literally delete the `isArchived` method. It is not used in a meaningful way.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14950:
URL: https://github.com/apache/flink/pull/14950#issuecomment-780431068
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13398",
"triggerID" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"triggerType" : "PUSH"
}, {
"hash" : "bee61356ebeb5f99c96ce3193b3bd8205e61e44d",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13409",
"triggerID" : "bee61356ebeb5f99c96ce3193b3bd8205e61e44d",
"triggerType" : "PUSH"
}, {
"hash" : "1b43a411d22af19f12f14a79c1094c8dc5cfeb42",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "1b43a411d22af19f12f14a79c1094c8dc5cfeb42",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* bee61356ebeb5f99c96ce3193b3bd8205e61e44d Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13409)
* 1b43a411d22af19f12f14a79c1094c8dc5cfeb42 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14950:
URL: https://github.com/apache/flink/pull/14950#issuecomment-780431068
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13398",
"triggerID" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"triggerType" : "PUSH"
}, {
"hash" : "bee61356ebeb5f99c96ce3193b3bd8205e61e44d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13409",
"triggerID" : "bee61356ebeb5f99c96ce3193b3bd8205e61e44d",
"triggerType" : "PUSH"
}, {
"hash" : "1b43a411d22af19f12f14a79c1094c8dc5cfeb42",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "1b43a411d22af19f12f14a79c1094c8dc5cfeb42",
"triggerType" : "PUSH"
}, {
"hash" : "e6aae42cf093d4f44b93b79a78bdc64538df8725",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13445",
"triggerID" : "e6aae42cf093d4f44b93b79a78bdc64538df8725",
"triggerType" : "PUSH"
}, {
"hash" : "cccca8191e6800961467c726695c8ef7575d157a",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13449",
"triggerID" : "cccca8191e6800961467c726695c8ef7575d157a",
"triggerType" : "PUSH"
}, {
"hash" : "44ee00afbd5ab0c81c53ed5787e7c0701b711e41",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "44ee00afbd5ab0c81c53ed5787e7c0701b711e41",
"triggerType" : "PUSH"
}, {
"hash" : "8298e7dcbbd9809429dd4c2d4148ff5e46bdb43b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13499",
"triggerID" : "8298e7dcbbd9809429dd4c2d4148ff5e46bdb43b",
"triggerType" : "PUSH"
}, {
"hash" : "d77a6c96e6616c2485ed787b7a1f0fedea97511b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13599",
"triggerID" : "d77a6c96e6616c2485ed787b7a1f0fedea97511b",
"triggerType" : "PUSH"
}, {
"hash" : "c1274ae69962c108df47f9939b3454154724acbb",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14049",
"triggerID" : "c1274ae69962c108df47f9939b3454154724acbb",
"triggerType" : "PUSH"
}, {
"hash" : "03ac0474397f9055837dc5727d634ac42ee18ce7",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14071",
"triggerID" : "03ac0474397f9055837dc5727d634ac42ee18ce7",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1b43a411d22af19f12f14a79c1094c8dc5cfeb42 UNKNOWN
* 44ee00afbd5ab0c81c53ed5787e7c0701b711e41 UNKNOWN
* c1274ae69962c108df47f9939b3454154724acbb Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14049)
* 03ac0474397f9055837dc5727d634ac42ee18ce7 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=14071)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] rmetzger commented on pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
rmetzger commented on pull request #14950:
URL: https://github.com/apache/flink/pull/14950#issuecomment-782112541
I've introduced another interface for the internal ExecutionGraph communication. Most likely, you'll have better names, but in general, the change feels much better 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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14950:
URL: https://github.com/apache/flink/pull/14950#issuecomment-780426957
Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.
## Automated Checks
Last check on commit cccca8191e6800961467c726695c8ef7575d157a (Fri Feb 19 07:34:50 UTC 2021)
**Warnings:**
* No documentation files were touched! Remember to keep the Flink docs up to date!
<sub>Mention the bot in a comment to re-run the automated checks.</sub>
## Review Progress
* ❓ 1. The [description] looks good.
* ❓ 2. There is [consensus] that the contribution should go into to Flink.
* ❓ 3. Needs [attention] from.
* ❓ 4. The change fits into the overall [architecture].
* ❓ 5. Overall code [quality] is good.
Please see the [Pull Request Review Guide](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process.<details>
The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required <summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot approve description` to approve one or more aspects (aspects: `description`, `consensus`, `architecture` and `quality`)
- `@flinkbot approve all` to approve all aspects
- `@flinkbot approve-until architecture` to approve everything until `architecture`
- `@flinkbot attention @username1 [@username2 ..]` to require somebody's attention
- `@flinkbot disapprove architecture` to remove an approval you gave earlier
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14950: [FLINK-21347][coordination] Extract interface from ExecutionGraph
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #14950:
URL: https://github.com/apache/flink/pull/14950#issuecomment-780431068
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13398",
"triggerID" : "e44c9cdb2e60872179635616e8ffb184bb0337b9",
"triggerType" : "PUSH"
}, {
"hash" : "bee61356ebeb5f99c96ce3193b3bd8205e61e44d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13409",
"triggerID" : "bee61356ebeb5f99c96ce3193b3bd8205e61e44d",
"triggerType" : "PUSH"
}, {
"hash" : "1b43a411d22af19f12f14a79c1094c8dc5cfeb42",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "1b43a411d22af19f12f14a79c1094c8dc5cfeb42",
"triggerType" : "PUSH"
}, {
"hash" : "e6aae42cf093d4f44b93b79a78bdc64538df8725",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13445",
"triggerID" : "e6aae42cf093d4f44b93b79a78bdc64538df8725",
"triggerType" : "PUSH"
}, {
"hash" : "cccca8191e6800961467c726695c8ef7575d157a",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13449",
"triggerID" : "cccca8191e6800961467c726695c8ef7575d157a",
"triggerType" : "PUSH"
}, {
"hash" : "44ee00afbd5ab0c81c53ed5787e7c0701b711e41",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "44ee00afbd5ab0c81c53ed5787e7c0701b711e41",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1b43a411d22af19f12f14a79c1094c8dc5cfeb42 UNKNOWN
* cccca8191e6800961467c726695c8ef7575d157a Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=13449)
* 44ee00afbd5ab0c81c53ed5787e7c0701b711e41 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org