You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ji...@apache.org on 2023/12/19 13:49:38 UTC
(flink) branch release-1.18 updated: [FLINK-33872] Retrieve checkpoint history for jobs in completed state
This is an automated email from the ASF dual-hosted git repository.
jingge pushed a commit to branch release-1.18
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.18 by this push:
new 54fd94ab750 [FLINK-33872] Retrieve checkpoint history for jobs in completed state
54fd94ab750 is described below
commit 54fd94ab750c6cbb4cfa21a911656235c1bb059b
Author: Hong Liang Teoh <li...@amazon.com>
AuthorDate: Mon Dec 18 11:39:13 2023 +0000
[FLINK-33872] Retrieve checkpoint history for jobs in completed state
---
.../flink/runtime/dispatcher/Dispatcher.java | 31 ++++----
.../flink/runtime/dispatcher/DispatcherTest.java | 84 ++++++++++++++++++++++
.../utils/ArchivedExecutionGraphBuilder.java | 10 ++-
3 files changed, 110 insertions(+), 15 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
index 7c9a92dfed4..cb54d11a966 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
@@ -905,29 +905,32 @@ public abstract class Dispatcher extends FencedRpcEndpoint<DispatcherId>
@Override
public CompletableFuture<ExecutionGraphInfo> requestExecutionGraphInfo(
JobID jobId, Time timeout) {
- Function<Throwable, ExecutionGraphInfo> checkExecutionGraphStoreOnException =
- throwable -> {
- // check whether it is a completed job
- final ExecutionGraphInfo executionGraphInfo =
- executionGraphInfoStore.get(jobId);
- if (executionGraphInfo == null) {
- throw new CompletionException(
- ExceptionUtils.stripCompletionException(throwable));
- } else {
- return executionGraphInfo;
- }
- };
Optional<JobManagerRunner> maybeJob = getJobManagerRunner(jobId);
return maybeJob.map(job -> job.requestJob(timeout))
.orElse(FutureUtils.completedExceptionally(new FlinkJobNotFoundException(jobId)))
- .exceptionally(checkExecutionGraphStoreOnException);
+ .exceptionally(t -> getExecutionGraphInfoFromStore(t, jobId));
+ }
+
+ private ExecutionGraphInfo getExecutionGraphInfoFromStore(Throwable t, JobID jobId) {
+ // check whether it is a completed job
+ final ExecutionGraphInfo executionGraphInfo = executionGraphInfoStore.get(jobId);
+ if (executionGraphInfo == null) {
+ throw new CompletionException(ExceptionUtils.stripCompletionException(t));
+ } else {
+ return executionGraphInfo;
+ }
}
@Override
public CompletableFuture<CheckpointStatsSnapshot> requestCheckpointStats(
JobID jobId, Time timeout) {
return performOperationOnJobMasterGateway(
- jobId, gateway -> gateway.requestCheckpointStats(timeout));
+ jobId, gateway -> gateway.requestCheckpointStats(timeout))
+ .exceptionally(
+ t ->
+ getExecutionGraphInfoFromStore(t, jobId)
+ .getArchivedExecutionGraph()
+ .getCheckpointStatsSnapshot());
}
@Override
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherTest.java
index d4f0ffc2b2e..391dbe15ccd 100755
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherTest.java
@@ -32,6 +32,7 @@ import org.apache.flink.core.testutils.OneShotLatch;
import org.apache.flink.runtime.blob.BlobServer;
import org.apache.flink.runtime.blob.PermanentBlobKey;
import org.apache.flink.runtime.checkpoint.CheckpointStatsSnapshot;
+import org.apache.flink.runtime.checkpoint.CheckpointStatsTracker;
import org.apache.flink.runtime.checkpoint.Checkpoints;
import org.apache.flink.runtime.checkpoint.metadata.CheckpointMetadata;
import org.apache.flink.runtime.client.DuplicateJobSubmissionException;
@@ -75,6 +76,7 @@ import org.apache.flink.runtime.messages.FlinkJobNotFoundException;
import org.apache.flink.runtime.messages.FlinkJobTerminatedWithoutCancellationException;
import org.apache.flink.runtime.messages.webmonitor.JobDetails;
import org.apache.flink.runtime.messages.webmonitor.MultipleJobsDetails;
+import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
import org.apache.flink.runtime.rest.handler.RestHandlerException;
import org.apache.flink.runtime.rest.handler.legacy.utils.ArchivedExecutionGraphBuilder;
import org.apache.flink.runtime.rpc.FatalErrorHandler;
@@ -683,6 +685,88 @@ public class DispatcherTest extends AbstractDispatcherTest {
Assertions.assertThat(resultsFuture).isCompletedWithValue(snapshot);
}
+ @Test
+ public void testRetrieveCheckpointStatsOnFailedJob() throws Exception {
+ testRetrieveCheckpointStatsWithJobStatus(JobStatus.FAILED);
+ }
+
+ @Test
+ public void testRetrieveCheckpointStatsOnFinishedJob() throws Exception {
+ testRetrieveCheckpointStatsWithJobStatus(JobStatus.FINISHED);
+ }
+
+ @Test
+ public void testRetrieveCheckpointStatsOnCancelledJob() throws Exception {
+ testRetrieveCheckpointStatsWithJobStatus(JobStatus.CANCELED);
+ }
+
+ private void testRetrieveCheckpointStatsWithJobStatus(JobStatus jobStatus) throws Exception {
+ CheckpointStatsSnapshot snapshot = getTestCheckpointStatsSnapshotWithTwoFailedCheckpoints();
+ TestingJobMasterGateway testingJobMasterGateway =
+ new TestingJobMasterGatewayBuilder().build();
+
+ dispatcher =
+ createAndStartDispatcher(
+ heartbeatServices,
+ haServices,
+ new TestingJobMasterGatewayJobManagerRunnerFactory(
+ testingJobMasterGateway));
+
+ ErrorInfo failureCause =
+ jobStatus == JobStatus.FAILED
+ ? new ErrorInfo(new RuntimeException("expected"), 1L)
+ : null;
+
+ final ExecutionGraphInfo completedExecutionGraphInfo =
+ new ExecutionGraphInfo(
+ new ArchivedExecutionGraphBuilder()
+ .setJobID(jobId)
+ .setState(jobStatus)
+ .setCheckpointStatsSnapshot(snapshot)
+ .setFailureCause(failureCause)
+ .build());
+ dispatcher.completeJobExecution(completedExecutionGraphInfo);
+
+ CompletableFuture<CheckpointStatsSnapshot> resultsFuture =
+ dispatcher.callAsyncInMainThread(
+ () -> dispatcher.requestCheckpointStats(jobId, TIMEOUT));
+ Assertions.assertThat(resultsFuture).succeedsWithin(Duration.ofSeconds(1));
+ Assertions.assertThat(resultsFuture).isCompletedWithValue(snapshot);
+ }
+
+ private CheckpointStatsSnapshot getTestCheckpointStatsSnapshotWithTwoFailedCheckpoints() {
+ CheckpointStatsTracker checkpointStatsTracker =
+ new CheckpointStatsTracker(
+ 10, UnregisteredMetricGroups.createUnregisteredTaskManagerMetricGroup());
+ checkpointStatsTracker.reportFailedCheckpointsWithoutInProgress();
+ checkpointStatsTracker.reportFailedCheckpointsWithoutInProgress();
+ return checkpointStatsTracker.createSnapshot();
+ }
+
+ @Test
+ public void testRetrieveCheckpointStatsOnNonExistentJob() throws Exception {
+ TestingJobMasterGateway testingJobMasterGateway =
+ new TestingJobMasterGatewayBuilder().build();
+
+ dispatcher =
+ createAndStartDispatcher(
+ heartbeatServices,
+ haServices,
+ new TestingJobMasterGatewayJobManagerRunnerFactory(
+ testingJobMasterGateway));
+
+ CompletableFuture<CheckpointStatsSnapshot> resultsFuture =
+ dispatcher.callAsyncInMainThread(
+ () -> dispatcher.requestCheckpointStats(jobId, TIMEOUT));
+
+ Assertions.assertThat(resultsFuture).failsWithin(Duration.ofSeconds(1));
+ Assertions.assertThat(resultsFuture).isCompletedExceptionally();
+
+ Assertions.assertThatThrownBy(resultsFuture::get)
+ .hasCauseInstanceOf(FlinkJobNotFoundException.class)
+ .hasMessageContaining("Could not find Flink job");
+ }
+
@Test
public void testThrowExceptionIfJobExecutionResultNotFound() throws Exception {
dispatcher =
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/utils/ArchivedExecutionGraphBuilder.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/utils/ArchivedExecutionGraphBuilder.java
index 4e074615361..32e567e4f65 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/utils/ArchivedExecutionGraphBuilder.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/utils/ArchivedExecutionGraphBuilder.java
@@ -22,6 +22,7 @@ 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.runtime.accumulators.StringifiedAccumulatorResult;
+import org.apache.flink.runtime.checkpoint.CheckpointStatsSnapshot;
import org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph;
import org.apache.flink.runtime.executiongraph.ArchivedExecutionJobVertex;
import org.apache.flink.runtime.executiongraph.ErrorInfo;
@@ -54,6 +55,7 @@ public class ArchivedExecutionGraphBuilder {
private ArchivedExecutionConfig archivedExecutionConfig;
private boolean isStoppable;
private Map<String, SerializedValue<OptionalFailure<Object>>> serializedUserAccumulators;
+ private CheckpointStatsSnapshot checkpointStatsSnapshot;
public ArchivedExecutionGraphBuilder setJobID(JobID jobID) {
this.jobID = jobID;
@@ -121,6 +123,12 @@ public class ArchivedExecutionGraphBuilder {
return this;
}
+ public ArchivedExecutionGraphBuilder setCheckpointStatsSnapshot(
+ CheckpointStatsSnapshot checkpointStatsSnapshot) {
+ this.checkpointStatsSnapshot = checkpointStatsSnapshot;
+ return this;
+ }
+
public ArchivedExecutionGraph build() {
JobID jobID = this.jobID != null ? this.jobID : new JobID();
String jobName = this.jobName != null ? this.jobName : "job_" + RANDOM.nextInt();
@@ -157,7 +165,7 @@ public class ArchivedExecutionGraphBuilder {
: new ArchivedExecutionConfigBuilder().build(),
isStoppable,
null,
- null,
+ checkpointStatsSnapshot,
"stateBackendName",
"checkpointStorageName",
TernaryBoolean.UNDEFINED,