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 2022/09/19 07:22:25 UTC
[GitHub] [flink] leletan opened a new pull request, #20852: feat(flink-runtime): add restful API to trigger checkpoints
leletan opened a new pull request, #20852:
URL: https://github.com/apache/flink/pull/20852
<!--
*Thank you very much for contributing to Apache Flink - we are happy that you want to help us improve Flink. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
*Please understand that we do not do this to make contributions to Flink a hassle. In order to uphold a high standard of quality for code contributions, while at the same time managing a large number of contributions, we need contributors to prepare the contributions well, and give reviewers enough contextual information for the review. Please also understand that contributions that do not follow this guide will take longer to review and thus typically be picked up with lower priority by the community.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [JIRA issue](https://issues.apache.org/jira/projects/FLINK/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no JIRA issue.
- Name the pull request in the form "[FLINK-XXXX] [component] Title of the pull request", where *FLINK-XXXX* should be replaced by the actual issue number. Skip *component* if you are unsure about which is the best component.
Typo fixes that have no associated JIRA issue should be named following this pattern: `[hotfix] [docs] Fix typo in event time introduction` or `[hotfix] [javadocs] Expand JavaDoc for PuncuatedWatermarkGenerator`.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes. You can set up Azure Pipelines CI to do that following [this guide](https://cwiki.apache.org/confluence/display/FLINK/Azure+Pipelines#AzurePipelines-Tutorial:SettingupAzurePipelinesforaforkoftheFlinkrepository).
- Each pull request should address only one issue, not mix up code from multiple issues.
- Each commit in the pull request has a meaningful commit message (including the JIRA id)
- Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
**(The sections below can be removed for hotfixes of typos)**
-->
## What is the purpose of the change
This pull request adds a restful API for one-time checkpointing triggering. This allows user to break incremental checkpoint chain by triggering a manual full checkpoint. Restful API delegates the scheduling to user.
## Brief change log
- The checkpoint trigger restful API is async, similar to savepoint restful API
- The checkpoint trigger API takes a CheckpointRetentionPolicy and will only trigger checkpoint once
- A new public triggerCheckpoint method taking a CheckpointProperty is added to CheckpointCoordinator
## Verifying this change
This change added tests and can be verified as follows:
- Added unit tests to validate CheckpointTriggerHandler and CheckpointTriggerStatusHandler restful handlers
- [WIP] Added integration tests for ...
- [WIP] Manually verified the changes by ...
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: no
- The serializers: no
- The runtime per-record code paths (performance sensitive): no
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn/Mesos, ZooKeeper: yes
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? yes
- If yes, how is the feature documented? [WIP]
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] leletan commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
leletan commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r994043353
##########
flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/checkpoints/CheckpointTriggerHandlers.java:
##########
@@ -0,0 +1,261 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.rest.handler.job.checkpoints;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.core.execution.CheckpointType;
+import org.apache.flink.runtime.dispatcher.UnknownOperationKeyException;
+import org.apache.flink.runtime.rest.handler.AbstractRestHandler;
+import org.apache.flink.runtime.rest.handler.HandlerRequest;
+import org.apache.flink.runtime.rest.handler.RestHandlerException;
+import org.apache.flink.runtime.rest.handler.async.AsynchronousOperationResult;
+import org.apache.flink.runtime.rest.handler.async.TriggerResponse;
+import org.apache.flink.runtime.rest.handler.job.AsynchronousJobOperationKey;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.JobIDPathParameter;
+import org.apache.flink.runtime.rest.messages.TriggerId;
+import org.apache.flink.runtime.rest.messages.TriggerIdPathParameter;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerHeaders;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerInfo;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerMessageParameters;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerRequestBody;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerStatusHeaders;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerStatusMessageParameters;
+import org.apache.flink.runtime.rpc.RpcUtils;
+import org.apache.flink.runtime.webmonitor.RestfulGateway;
+import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.SerializedThrowable;
+
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+import javax.annotation.Nonnull;
+
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+
+/**
+ * HTTP handlers for asynchronous triggering of checkpoints.
+ *
+ * <p>Drawing checkpoints is a potentially long-running operation. To avoid blocking HTTP
+ * connections, checkpoints must be drawn in two steps. First, an HTTP request is issued to trigger
+ * the checkpoint asynchronously. The request will be assigned a {@link TriggerId}, which is
+ * returned in the response body. Next, the returned {@link TriggerId} should be used to poll the
+ * status of the checkpoint until it is finished.
+ *
+ * <p>A checkpoint is triggered by sending an HTTP {@code POST} request to {@code
+ * /jobs/:jobid/checkpoints}. The HTTP request may contain a JSON body to specify a customized
+ * {@link TriggerId} and a {@link CheckpointType}, e.g.,
+ *
+ * <pre>
+ * { "triggerId": "7d273f5a62eb4730b9dea8e833733c1e", "checkpointType": "FULL" }
+ * </pre>
+ *
+ * <p>If the body is omitted, or the field {@code checkpointType} is {@code null}, the default
+ * checkpointType as specified by {@link CheckpointType#CONFIGURED} will be used. As written above,
+ * the response will contain a request id, e.g.,
+ *
+ * <pre>
+ * { "request-id": "7d273f5a62eb4730b9dea8e833733c1e" }
+ * </pre>
+ *
+ * <p>To poll for the status of an ongoing checkpoint, an HTTP {@code GET} request is issued to
+ * {@code /jobs/:jobid/checkpoints/status/:checkpointtriggerid}. If the specified checkpoint is
+ * still ongoing, the response will be
+ *
+ * <pre>
+ * {
+ * "status": {
+ * "id": "IN_PROGRESS"
+ * }
+ * }
+ * </pre>
+ *
+ * <p>If the specified checkpoints has completed, the status id will transition to {@code
+ * COMPLETED}, and the response will additionally contain information about the savepoint, such as
+ * the location:
+ *
+ * <pre>
+ * {
+ * "status": {
+ * "id": "COMPLETED"
+ * },
+ * "operation": {
+ * "checkpointId": "123"
+ * }
+ * }
+ * </pre>
+ */
+public class CheckpointTriggerHandlers {
+
+ /** handler for the checkpoint trigger operation. */
+ public static class CheckpointTriggerHandler
+ extends AbstractRestHandler<
+ RestfulGateway,
+ CheckpointTriggerRequestBody,
+ TriggerResponse,
+ CheckpointTriggerMessageParameters> {
+
+ public CheckpointTriggerHandler(
+ final GatewayRetriever<? extends RestfulGateway> leaderRetriever,
+ final Time timeout,
+ final Map<String, String> responseHeaders) {
+ super(
+ leaderRetriever,
+ timeout,
+ responseHeaders,
+ CheckpointTriggerHeaders.getInstance());
+ }
+
+ protected AsynchronousJobOperationKey createOperationKey(
+ final HandlerRequest<CheckpointTriggerRequestBody> request) {
+ final JobID jobId = request.getPathParameter(JobIDPathParameter.class);
+ return AsynchronousJobOperationKey.of(
+ request.getRequestBody().getTriggerId().orElseGet(TriggerId::new), jobId);
+ }
+
+ @Override
+ protected CompletableFuture<TriggerResponse> handleRequest(
+ @Nonnull HandlerRequest<CheckpointTriggerRequestBody> request,
+ @Nonnull RestfulGateway gateway)
+ throws RestHandlerException {
+ final AsynchronousJobOperationKey operationKey = createOperationKey(request);
+
+ return gateway.triggerCheckpoint(
+ operationKey,
+ request.getRequestBody().getCheckpointType(),
+ RpcUtils.INF_TIMEOUT)
Review Comment:
Created [this](https://issues.apache.org/jira/browse/FLINK-29610) JIRA.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] stevenzwu commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r981949709
##########
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java:
##########
@@ -511,6 +514,42 @@ public CompletableFuture<CompletedCheckpoint> triggerCheckpoint(boolean isPeriod
return triggerCheckpointFromCheckpointThread(checkpointProperties, null, isPeriodic);
}
+ /**
+ * Triggers one new checkpoint with the given checkpointType. If the given checkpointType is
+ * null, then it will fall back to use the CheckpointCoordinator's checkpointProperties. The
+ * return value is a future. It completes when the checkpoint triggered finishes or an error
+ * occurred.
+ *
+ * @param checkpointType specifies the back up type of the checkpoint to trigger.
+ * @return a future to the completed checkpoint.
+ */
+ public CompletableFuture<CompletedCheckpoint> triggerCheckpoint(
+ @Nullable CheckpointType checkpointType) {
+
+ final SnapshotType snapshotType;
+ if (checkpointType == null) {
+ snapshotType = checkpointProperties.getCheckpointType();
Review Comment:
we should rename the `getCheckpointType` method 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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] stevenzwu commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r980390428
##########
flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java:
##########
@@ -863,11 +871,6 @@ public CompletableFuture<String> triggerSavepoint(
return schedulerNG.triggerSavepoint(targetDirectory, cancelJob, formatType);
}
- @Override
Review Comment:
got it. you added a default impl method in the parent interface.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] stevenzwu commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r981948068
##########
flink-core/src/main/java/org/apache/flink/core/execution/CheckpointType.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * 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.core.execution;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.DescribedEnum;
+import org.apache.flink.configuration.description.InlineElement;
+
+import static org.apache.flink.configuration.description.TextElement.text;
+
+/** Describes the type in which a checkpoint should be taken. */
+@PublicEvolving
+public enum CheckpointType implements DescribedEnum {
+ /** A checkpoint type with the entire state check pointed, common for all state backends. */
+ FULL("A checkpoint type with the entire state check pointed, common for all state backends."),
+
+ /**
+ * A checkpoint type that check points only the difference between snapshots, specific for
Review Comment:
the Javadoc seems unnecessary considering the description already in place.
##########
flink-core/src/main/java/org/apache/flink/core/execution/CheckpointType.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * 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.core.execution;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.DescribedEnum;
+import org.apache.flink.configuration.description.InlineElement;
+
+import static org.apache.flink.configuration.description.TextElement.text;
+
+/** Describes the type in which a checkpoint should be taken. */
+@PublicEvolving
+public enum CheckpointType implements DescribedEnum {
+ /** A checkpoint type with the entire state check pointed, common for all state backends. */
Review Comment:
the Javadoc seems unnecessary considering the description already in place.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] leletan commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
leletan commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r988586135
##########
flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/checkpoints/CheckpointTriggerStatusHeaders.java:
##########
@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.rest.messages.checkpoints;
+
+import org.apache.flink.runtime.rest.HttpMethodWrapper;
+import org.apache.flink.runtime.rest.handler.async.AsynchronousOperationStatusMessageHeaders;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.JobIDPathParameter;
+import org.apache.flink.runtime.rest.messages.TriggerIdPathParameter;
+
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+/** These headers define the protocol for triggering a checkpoint. */
+public class CheckpointTriggerStatusHeaders
+ extends AsynchronousOperationStatusMessageHeaders<
+ CheckpointTriggerInfo, CheckpointTriggerStatusMessageParameters> {
+
+ private static final CheckpointTriggerStatusHeaders INSTANCE =
+ new CheckpointTriggerStatusHeaders();
+
+ private static final String URL =
+ String.format(
+ "/jobs/:%s/checkpoints/status/:%s",
+ JobIDPathParameter.KEY, TriggerIdPathParameter.KEY);
+
+ private CheckpointTriggerStatusHeaders() {}
+
+ @Override
+ public Class<EmptyRequestBody> getRequestClass() {
+ return EmptyRequestBody.class;
+ }
+
+ @Override
+ public HttpResponseStatus getResponseStatusCode() {
+ return HttpResponseStatus.OK;
+ }
+
+ @Override
+ public CheckpointTriggerStatusMessageParameters getUnresolvedMessageParameters() {
+ return new CheckpointTriggerStatusMessageParameters();
+ }
+
+ @Override
+ public HttpMethodWrapper getHttpMethod() {
+ return HttpMethodWrapper.GET;
+ }
+
+ @Override
+ public String getTargetRestEndpointURL() {
+ return URL;
+ }
+
+ public static CheckpointTriggerStatusHeaders getInstance() {
Review Comment:
good catch, will fix
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] stevenzwu commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r980387378
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerBase.java:
##########
@@ -878,7 +879,8 @@ public CompletableFuture<String> triggerSavepoint(
}
@Override
- public CompletableFuture<String> triggerCheckpoint() {
+ public CompletableFuture<CompletedCheckpoint> triggerCheckpoint(
Review Comment:
oh. I didn't realize it is needed in the trigger checkpoint code path.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] stevenzwu commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r979055803
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerBase.java:
##########
@@ -878,7 +879,8 @@ public CompletableFuture<String> triggerSavepoint(
}
@Override
- public CompletableFuture<String> triggerCheckpoint() {
+ public CompletableFuture<CompletedCheckpoint> triggerCheckpoint(
Review Comment:
we probably don't need to change scheduler (for periodical checkpoint).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] stevenzwu commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r981948317
##########
flink-core/src/main/java/org/apache/flink/core/execution/CheckpointType.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * 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.core.execution;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.DescribedEnum;
+import org.apache.flink.configuration.description.InlineElement;
+
+import static org.apache.flink.configuration.description.TextElement.text;
+
+/** Describes the type in which a checkpoint should be taken. */
+@PublicEvolving
+public enum CheckpointType implements DescribedEnum {
+ /** A checkpoint type with the entire state check pointed, common for all state backends. */
+ FULL("A checkpoint type with the entire state check pointed, common for all state backends."),
+
+ /**
+ * A checkpoint type that check points only the difference between snapshots, specific for
+ * certain state backend.
+ */
+ INCREMENTAL(
+ "A checkpoint type that check points only the difference between snapshots, specific for certain state backend.");
Review Comment:
should be one word: check points -> checkpoints.
Also I see the style difference in the description btw full and incremental. We probably can reword the description for full as following.
```
A checkpoint type that checkpoints the entire state, ...
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] leletan commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
leletan commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r974889748
##########
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointRetentionPolicy.java:
##########
@@ -24,6 +24,9 @@
@Internal
public enum CheckpointRetentionPolicy {
+ /** Full Checkpoints should be retained on cancellation and failure. */
+ FULL_RETAIN_ON_CANCELLATION,
Review Comment:
Good catch @pnowojski. This was one of the place I was not very confident with. As a hindsight, totally agree it was a bad design.
The reason why I did that was as following:
I was not sure if a CheckpointProperty will be valid if it is too far away from [these](https://github.com/apache/flink/blob/d8a4304b892412eab4c5c19b5deb84166943d3bb/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointProperties.java#L246) pre-configured ones, so I was thinking of creating one as close to an existing pre-configured as possible. Meanwhile the public method to create a pre-configured CheckpointProperty was through [this](https://github.com/apache/flink/blob/d8a4304b892412eab4c5c19b5deb84166943d3bb/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointProperties.java#L349) method with a `CheckpointRetentionPolicy`.
Meanwhile I would love to hear your thoughts
1. What checkpoint configurations we should expose in the restful API? Are there more general use cases requesting other configurations for checkpoint manual triggering (other than ours)?
2. Or would it be better just to start with our specific use case - always only trigger a full checkpoint, i.e. reuse all `CheckpointProperty` and only replace `CheckpointType` with [`FULL_CHECKPOINT `](https://github.com/apache/flink/blob/d8a4304b892412eab4c5c19b5deb84166943d3bb/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointType.java#L30)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] leletan commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
leletan commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r974889748
##########
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointRetentionPolicy.java:
##########
@@ -24,6 +24,9 @@
@Internal
public enum CheckpointRetentionPolicy {
+ /** Full Checkpoints should be retained on cancellation and failure. */
+ FULL_RETAIN_ON_CANCELLATION,
Review Comment:
Good catch @pnowojski. This was one of the place I was not very confident with. As a hindsight, totally agree it was a bad design.
The reason why I did that was as following:
I was not sure if a CheckpointProperty will be valid if it is too far away from [these](https://github.com/apache/flink/blob/d8a4304b892412eab4c5c19b5deb84166943d3bb/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointProperties.java#L246) pre-configured ones so I was thinking of creating one as close to an existing one as possible. Meanwhile the public method to create a pre-configured CheckpointProperty was through [this](https://github.com/apache/flink/blob/d8a4304b892412eab4c5c19b5deb84166943d3bb/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointProperties.java#L349) method with a `CheckpointRetentionPolicy`.
Meanwhile I would love to hear your thoughts
1. What checkpoint configurations we should expose in the restful API? Are there more general use cases requesting other configurations for checkpoint manual triggering (other than ours)?
2. Or would it be better just to start with our specific use case - always only trigger a full checkpoint, i.e. reuse all `CheckpointProperty` and only replace `CheckpointType` with [`FULL_CHECKPOINT `](https://github.com/apache/flink/blob/d8a4304b892412eab4c5c19b5deb84166943d3bb/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointType.java#L30)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] leletan commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
leletan commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r974889748
##########
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointRetentionPolicy.java:
##########
@@ -24,6 +24,9 @@
@Internal
public enum CheckpointRetentionPolicy {
+ /** Full Checkpoints should be retained on cancellation and failure. */
+ FULL_RETAIN_ON_CANCELLATION,
Review Comment:
Good catch @pnowojski. This was one of the places I was not very confident with. As a hindsight, totally agree it was a bad design.
The reason why I did that was as following:
I was not sure if a CheckpointProperty will be valid if it is too far away from [these](https://github.com/apache/flink/blob/d8a4304b892412eab4c5c19b5deb84166943d3bb/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointProperties.java#L246) pre-configured ones, so I was thinking of creating one as close to an existing pre-configured as possible. Meanwhile the public method to create a pre-configured CheckpointProperty was through [this](https://github.com/apache/flink/blob/d8a4304b892412eab4c5c19b5deb84166943d3bb/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointProperties.java#L349) method with a `CheckpointRetentionPolicy`. And I did not notice the `@Internal` for the `CheckpointRetentionPolicy` just saw it was `public`.
Meanwhile I would love to hear your thoughts
1. What checkpoint configurations we should expose in the restful API? Are there more general use cases requesting other configurations for checkpoint manual triggering (other than ours)?
2. Or would it be better just to start with our specific use case - always only trigger a full checkpoint, i.e. reuse all `CheckpointProperty` and only replace `CheckpointType` with [`FULL_CHECKPOINT `](https://github.com/apache/flink/blob/d8a4304b892412eab4c5c19b5deb84166943d3bb/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointType.java#L30)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] leletan commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
leletan commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r980350376
##########
flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java:
##########
@@ -863,11 +871,6 @@ public CompletableFuture<String> triggerSavepoint(
return schedulerNG.triggerSavepoint(targetDirectory, cancelJob, formatType);
}
- @Override
Review Comment:
Double checked the code - the backward compatibility should already been kept. At this class's parent interface `JobMasterGateway`, this method is supported as
```
/**
* Triggers taking a checkpoint of the executed job.
*
* @param timeout for the rpc call
* @return Future which is completed with the checkpoint path once completed
*/
default CompletableFuture<String> triggerCheckpoint(@RpcTimeout final Time timeout) {
return triggerCheckpoint(null, timeout).thenApply(CompletedCheckpoint::getExternalPointer);
}
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] leletan commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
leletan commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r989292326
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerNG.java:
##########
@@ -125,7 +127,8 @@ void notifyKvStateUnregistered(
CompletableFuture<String> triggerSavepoint(
@Nullable String targetDirectory, boolean cancelJob, SavepointFormatType formatType);
- CompletableFuture<String> triggerCheckpoint();
+ CompletableFuture<CompletedCheckpoint> triggerCheckpoint(
+ @Nullable CheckpointType checkpointType);
Review Comment:
Good point. Let me see if I can do this without breaking any public API
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] leletan commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
leletan commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r984896360
##########
flink-core/src/main/java/org/apache/flink/core/execution/CheckpointType.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * 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.core.execution;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.DescribedEnum;
+import org.apache.flink.configuration.description.InlineElement;
+
+import static org.apache.flink.configuration.description.TextElement.text;
+
+/** Describes the type in which a checkpoint should be taken. */
+@PublicEvolving
+public enum CheckpointType implements DescribedEnum {
+ /** A checkpoint type with the entire state check pointed, common for all state backends. */
Review Comment:
This is copied from [SavepointFormatType](https://github.com/apache/flink/blob/f543b8ac690b1dee58bc3cb345a1c8ad0db0941e/flink-core/src/main/java/org/apache/flink/core/execution/SavepointFormatType.java)
But I am fine with removing it.
##########
flink-core/src/main/java/org/apache/flink/core/execution/CheckpointType.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * 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.core.execution;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.DescribedEnum;
+import org.apache.flink.configuration.description.InlineElement;
+
+import static org.apache.flink.configuration.description.TextElement.text;
+
+/** Describes the type in which a checkpoint should be taken. */
+@PublicEvolving
+public enum CheckpointType implements DescribedEnum {
+ /** A checkpoint type with the entire state check pointed, common for all state backends. */
+ FULL("A checkpoint type with the entire state check pointed, common for all state backends."),
+
+ /**
+ * A checkpoint type that check points only the difference between snapshots, specific for
Review Comment:
This is copied from [SavepointFormatType](https://github.com/apache/flink/blob/f543b8ac690b1dee58bc3cb345a1c8ad0db0941e/flink-core/src/main/java/org/apache/flink/core/execution/SavepointFormatType.java)
But I am fine with removing 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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] leletan commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
leletan commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r984896568
##########
flink-core/src/main/java/org/apache/flink/core/execution/CheckpointType.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * 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.core.execution;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.DescribedEnum;
+import org.apache.flink.configuration.description.InlineElement;
+
+import static org.apache.flink.configuration.description.TextElement.text;
+
+/** Describes the type in which a checkpoint should be taken. */
+@PublicEvolving
+public enum CheckpointType implements DescribedEnum {
+ /** A checkpoint type with the entire state check pointed, common for all state backends. */
+ FULL("A checkpoint type with the entire state check pointed, common for all state backends."),
+
+ /**
+ * A checkpoint type that check points only the difference between snapshots, specific for
+ * certain state backend.
+ */
+ INCREMENTAL(
+ "A checkpoint type that check points only the difference between snapshots, specific for certain state backend.");
+ private final InlineElement description;
+ public static final CheckpointType DEFAULT = CheckpointType.FULL;
+
+ CheckpointType(String description) {
+ this.description = text(description);
+ }
+
+ @Override
+ @Internal
Review Comment:
This is copied from [SavepointFormatType](https://github.com/apache/flink/blob/f543b8ac690b1dee58bc3cb345a1c8ad0db0941e/flink-core/src/main/java/org/apache/flink/core/execution/SavepointFormatType.java#L50)
But I am fine with removing 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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] leletan commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
leletan commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r974874914
##########
flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterGateway.java:
##########
@@ -218,13 +220,27 @@ CompletableFuture<String> triggerSavepoint(
final SavepointFormatType formatType,
@RpcTimeout final Time timeout);
+ /**
+ * Triggers taking a checkpoint of the executed job.
+ *
+ * @param checkpointProperties to determine how checkpoint should be taken or null if the
+ * existing checkpoint property should be used
+ * @param timeout for the rpc call
+ * @return Future which is completed with the savepoint path once completed
Review Comment:
Good catch
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] leletan commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
leletan commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r974889748
##########
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointRetentionPolicy.java:
##########
@@ -24,6 +24,9 @@
@Internal
public enum CheckpointRetentionPolicy {
+ /** Full Checkpoints should be retained on cancellation and failure. */
+ FULL_RETAIN_ON_CANCELLATION,
Review Comment:
Good catch @pnowojski. This was one of the place I was not very confident with. As a hindsight, totally agree it was a bad design.
The reason why I did that was as following:
I was not sure if a CheckpointProperty will be valid if it is too far away from [these](https://github.com/apache/flink/blob/d8a4304b892412eab4c5c19b5deb84166943d3bb/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointProperties.java#L246) pre-configured ones, so I was thinking of creating one as close to an existing pre-configured as possible. Meanwhile the public method to create a pre-configured CheckpointProperty was through [this](https://github.com/apache/flink/blob/d8a4304b892412eab4c5c19b5deb84166943d3bb/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointProperties.java#L349) method with a `CheckpointRetentionPolicy`. And I did not notice the `@Internal` for the `CheckpointRetentionPolicy` just saw it was `public`.
Meanwhile I would love to hear your thoughts
1. What checkpoint configurations we should expose in the restful API? Are there more general use cases requesting other configurations for checkpoint manual triggering (other than ours)?
2. Or would it be better just to start with our specific use case - always only trigger a full checkpoint, i.e. reuse all `CheckpointProperty` and only replace `CheckpointType` with [`FULL_CHECKPOINT `](https://github.com/apache/flink/blob/d8a4304b892412eab4c5c19b5deb84166943d3bb/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointType.java#L30)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] leletan commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
leletan commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r974870165
##########
flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterGateway.java:
##########
@@ -218,13 +220,27 @@ CompletableFuture<String> triggerSavepoint(
final SavepointFormatType formatType,
@RpcTimeout final Time timeout);
+ /**
+ * Triggers taking a checkpoint of the executed job.
+ *
+ * @param checkpointProperties to determine how checkpoint should be taken or null if the
+ * existing checkpoint property should be used
+ * @param timeout for the rpc call
+ * @return Future which is completed with the savepoint path once completed
+ */
+ CompletableFuture<CompletedCheckpoint> triggerCheckpoint(
Review Comment:
Good question, @pnowojski.
Assuming we are talking about
1. `CompletableFuture<String> triggerCheckpoint(@RpcTimeout final Time timeout)` as the other method
2. `CompletedOperationCache<AsynchronousJobOperationKey, Long> checkpointTriggerCache` in the `DispatcherCachedOperationsHandler` as the cache (location as cache value?)
We definitely can. Meanwhile I guess this means that the restful API (checkpoint trigger status API) will eventually return checkpoint location in the restful response body, instead of checkpoint id.
The reason for returning `CompletedCheckpoint` was to facilitate both old method and new API: the new restful API was returning checkpoint id, while the other method was returning checkpoint location, and CompletedCheckpoint contains both.
I am pretty new to flink so I am not sure what are people typically care about a checkpoint.
If we think it is a better design to return checkpoint location, instead of checkpoint id in the response body in the restful API, I am happy to return checkpoint location and reuse existing cache and another method.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] stevenzwu commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r979031935
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DispatcherCachedOperationsHandler.java:
##########
@@ -39,30 +40,66 @@ public class DispatcherCachedOperationsHandler {
private final CompletedOperationCache<AsynchronousJobOperationKey, String>
savepointTriggerCache;
+ private final CompletedOperationCache<AsynchronousJobOperationKey, Long> checkpointTriggerCache;
+
+ private final TriggerCheckpointFunction triggerCheckpointFunction;
+
private final TriggerSavepointFunction triggerSavepointFunction;
private final TriggerSavepointFunction stopWithSavepointFunction;
DispatcherCachedOperationsHandler(
DispatcherOperationCaches operationCaches,
+ TriggerCheckpointFunction triggerCheckpointFunction,
TriggerSavepointFunction triggerSavepointFunction,
TriggerSavepointFunction stopWithSavepointFunction) {
this(
+ triggerCheckpointFunction,
+ operationCaches.getCheckpointTriggerCache(),
triggerSavepointFunction,
stopWithSavepointFunction,
operationCaches.getSavepointTriggerCache());
}
@VisibleForTesting
DispatcherCachedOperationsHandler(
+ TriggerCheckpointFunction triggerCheckpointFunction,
+ CompletedOperationCache<AsynchronousJobOperationKey, Long> checkpointTriggerCache,
TriggerSavepointFunction triggerSavepointFunction,
TriggerSavepointFunction stopWithSavepointFunction,
CompletedOperationCache<AsynchronousJobOperationKey, String> savepointTriggerCache) {
+ this.triggerCheckpointFunction = triggerCheckpointFunction;
+ this.checkpointTriggerCache = checkpointTriggerCache;
this.triggerSavepointFunction = triggerSavepointFunction;
this.stopWithSavepointFunction = stopWithSavepointFunction;
this.savepointTriggerCache = savepointTriggerCache;
}
+ public CompletableFuture<Acknowledge> triggerCheckpoint(
+ AsynchronousJobOperationKey operationKey,
+ CheckpointBackupType checkpointBackupType,
+ Time timeout) {
+
+ if (!checkpointTriggerCache.containsOperation(operationKey)) {
+ checkpointTriggerCache.registerOngoingOperation(
+ operationKey,
+ triggerCheckpointFunction.apply(
+ operationKey.getJobId(), checkpointBackupType, timeout));
+ }
+
+ return CompletableFuture.completedFuture(Acknowledge.get());
+ }
+
+ public CompletableFuture<OperationResult<Long>> getCheckpointStatus(
Review Comment:
The result type is `Long`. what does the long value result indicate? I assume the `Long` type is not meant for checkpoint Id.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] stevenzwu commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r979033459
##########
flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java:
##########
@@ -853,6 +855,12 @@ public CompletableFuture<ExecutionGraphInfo> requestJob(Time timeout) {
return CompletableFuture.completedFuture(schedulerNG.requestJob());
}
+ @Override
+ public CompletableFuture<CompletedCheckpoint> triggerCheckpoint(
Review Comment:
nit: this can be added next to the other `triggerCheckpoint` method
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] leletan commented on pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
leletan commented on PR #20852:
URL: https://github.com/apache/flink/pull/20852#issuecomment-1263952346
As per discussion https://github.com/apache/flink/pull/20852#issuecomment-1260402519, will address the internal class renaming in a follow up PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] stevenzwu commented on pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on PR #20852:
URL: https://github.com/apache/flink/pull/20852#issuecomment-1264009972
@leletan we shall come back to this PR after the internal class renaming is merged. Please create a jira and PR for that first.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] leletan commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
leletan commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r988641876
##########
flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/checkpoints/CheckpointTriggerRequestBody.java:
##########
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.rest.messages.checkpoints;
+
+import org.apache.flink.core.execution.CheckpointType;
+import org.apache.flink.runtime.rest.messages.RequestBody;
+import org.apache.flink.runtime.rest.messages.TriggerId;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnore;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import javax.annotation.Nullable;
+
+import java.util.Optional;
+
+/** {@link RequestBody} to trigger checkpoints. */
+public class CheckpointTriggerRequestBody implements RequestBody {
+
+ private static final String FIELD_NAME_TRIGGER_ID = "triggerId";
+ private static final String FIELD_NAME_CHECKPOINT_TYPE = "checkpointType";
Review Comment:
Oops, good catch.
There were some code changes after generating the doc based on PR comments, forgot to update doc. Will update docs.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] stevenzwu commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r981947213
##########
flink-core/src/main/java/org/apache/flink/core/execution/CheckpointType.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * 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.core.execution;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.DescribedEnum;
+import org.apache.flink.configuration.description.InlineElement;
+
+import static org.apache.flink.configuration.description.TextElement.text;
+
+/** Describes the type in which a checkpoint should be taken. */
+@PublicEvolving
+public enum CheckpointType implements DescribedEnum {
+ /** A checkpoint type with the entire state check pointed, common for all state backends. */
+ FULL("A checkpoint type with the entire state check pointed, common for all state backends."),
Review Comment:
should be one world: check pointed -> checkpointed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] leletan commented on pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
leletan commented on PR #20852:
URL: https://github.com/apache/flink/pull/20852#issuecomment-1265689101
Thanks, @pnowojski and @stevenzwu!
Combing the feedbacks and here is my plan:
1. fix the build
2. rebase all the changes so far into one commit.
3. do a follow up commit for the class renaming in this PR
Let me know if you have concern about above.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] leletan commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
leletan commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r988446731
##########
flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/checkpoints/CheckpointTriggerInfo.java:
##########
@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.rest.messages.checkpoints;
+
+import org.apache.flink.runtime.rest.messages.ResponseBody;
+import org.apache.flink.runtime.rest.messages.json.SerializedThrowableDeserializer;
+import org.apache.flink.runtime.rest.messages.json.SerializedThrowableSerializer;
+import org.apache.flink.util.SerializedThrowable;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.annotation.JsonSerialize;
+
+import javax.annotation.Nullable;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+
+/** Represents information about a finished triggered checkpoint. */
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public class CheckpointTriggerInfo implements ResponseBody {
+
+ private static final String FIELD_NAME_CHECKPOINT_ID = "checkpointId";
+
+ private static final String FIELD_NAME_FAILURE_CAUSE = "failure-cause";
+
+ @JsonProperty(FIELD_NAME_CHECKPOINT_ID)
+ @Nullable
+ private final Long checkpointId;
+
+ @JsonProperty(FIELD_NAME_FAILURE_CAUSE)
+ @JsonSerialize(using = SerializedThrowableSerializer.class)
+ @JsonDeserialize(using = SerializedThrowableDeserializer.class)
+ @Nullable
+ private final SerializedThrowable failureCause;
+
+ @JsonCreator
+ public CheckpointTriggerInfo(
+ @JsonProperty(FIELD_NAME_CHECKPOINT_ID) @Nullable final Long checkpointId,
+ @JsonProperty(FIELD_NAME_FAILURE_CAUSE)
+ @JsonDeserialize(using = SerializedThrowableDeserializer.class)
+ @Nullable
+ final SerializedThrowable failureCause) {
+ checkArgument(
+ checkpointId != null ^ failureCause != null,
+ "Either checkpointId or failureCause must be set");
+
+ this.checkpointId = checkpointId;
+ this.failureCause = failureCause;
+ }
+
+ @Nullable
+ public Long getCheckpointId() {
+ return checkpointId;
+ }
+
+ @Nullable
+ public SerializedThrowable getFailureCause() {
+ return failureCause;
+ }
Review Comment:
Sure. Do we want to fix [SavepointInfo](https://github.com/apache/flink/blob/99c2a415e9eeefafacf70762b6f54070f7911ceb/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/savepoints/SavepointInfo.java#L74) as well?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zentol commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
zentol commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r993242999
##########
flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/checkpoints/CheckpointTriggerStatusHeaders.java:
##########
@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.rest.messages.checkpoints;
+
+import org.apache.flink.runtime.rest.HttpMethodWrapper;
+import org.apache.flink.runtime.rest.handler.async.AsynchronousOperationStatusMessageHeaders;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.JobIDPathParameter;
+import org.apache.flink.runtime.rest.messages.TriggerIdPathParameter;
+
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+/** These headers define the protocol for triggering a checkpoint. */
+public class CheckpointTriggerStatusHeaders
+ extends AsynchronousOperationStatusMessageHeaders<
+ CheckpointTriggerInfo, CheckpointTriggerStatusMessageParameters> {
Review Comment:
> this is very similar to savepoint logic, which has no "trigger" for its API or class names.
That's why I'd like avoid it. It behaves the same way but currently has a different naming scheme.
> 1. The new APIs in this PR is for manual trigger of a checkpoint and it may or may not end up with a checkpoint (it could end up with a failure and return with a failure cause, before it can reach CheckpointCoordinator)
That's a good observation but I'd say it's fine nonetheless to not refer to "triggers".
It's just a technical limitation that we can't differentiate between failures before/after the checkpoint/savepoint was triggered that we _may_ fix at some point.
I also think this distinction isn't really important. It would of course be great if the initial request could already fail early, but at the end of the day the results are the same to the user.
> 2. The checkpoint trigger status is queried by trigger id and its value is read from DispatcherOperationCaches which has only 5 min cache duration by default, so it is more of a trigger status than checkpoint status.
The timeout is there because the JM waits with it's shutdown until the result of these requests have been consumed, and we don't want it to sit there idling indefinitely because someone forget to consume a checkpoint result.
> 3. There are already other APIs for querying checkpoint metadata and details, which should cover all the retained checkpoints for the application - don't want to confuse user with them.
Interesting point. However this also applies to the savepoints API and so far no one complained. I think there's a well-defined use-cases for both checkpoint-related APIs; one gives you the full details, the other just status information about manually triggered checkpoints (which is somewhat explicit since the url has a triggerId in it).
I wouldn't worry about 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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] leletan commented on pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
leletan commented on PR #20852:
URL: https://github.com/apache/flink/pull/20852#issuecomment-1265618274
@flinkbot run azure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
pnowojski commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r975199952
##########
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointRetentionPolicy.java:
##########
@@ -24,6 +24,9 @@
@Internal
public enum CheckpointRetentionPolicy {
+ /** Full Checkpoints should be retained on cancellation and failure. */
+ FULL_RETAIN_ON_CANCELLATION,
Review Comment:
I could see use cases for manually triggering incremental checkpoints as well, so I would suggest to full/incremental parameter. We would need to decouple REST API constants from the internal classes and add a translation layer for that. It should be probably something similar to `org.apache.flink.core.execution.SavepointFormatType`. `CheckpointType` with two supported values `INCREMENTAL` and `FULL`.
As for the `CheckpointProperties` class, probably you should copy instead of passing `CheckpointProperties` in `org.apache.flink.runtime.checkpoint.CheckpointCoordinator#triggerCheckpoint(org.apache.flink.runtime.checkpoint.CheckpointProperties)`, you should just pass `org.apache.flink.runtime.checkpoint.CheckpointType` and inside the `triggerCheckpoint` construct new `CheckpointProperties` by copying `org.apache.flink.runtime.checkpoint.CheckpointCoordinator#checkpointProperties`. Something like that:
```
public CompletableFuture<CompletedCheckpoint> triggerCheckpoint(CheckpointType checkpointType) {
CheckpointProperties properties =
new CheckpointProperties(
checkpointProperties.forceCheckpoint(),
checkpointType,
checkpointProperties.discardOnSubsumed(),
checkpointProperties.discardOnJobFinished(),
checkpointProperties.discardOnJobCancelled(),
checkpointProperties.discardOnJobFailed(),
checkpointProperties.discardOnJobSuspended(),
checkpointProperties.isUnclaimed());
return triggerCheckpointFromCheckpointThread(properties, null, false);
}
```
Rest API Handler would then translate the REST API `CheckpointType` into the internal `CheckpointType` (note there would be two `CheckpointType`. One enum in the REST API, the other, pre existing `org.apache.flink.runtime.checkpoint.CheckpointType` class).
However it would be great if @zentol could validate if that's the way it should be done from the REST API perspective 😅
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] leletan commented on pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
leletan commented on PR #20852:
URL: https://github.com/apache/flink/pull/20852#issuecomment-1251824681
Also I am wondering if this change would need any further test coverages? If yes, where would be the best place I should start looking?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] stevenzwu commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r979072658
##########
flink-core/src/main/java/org/apache/flink/core/execution/CheckpointBackupType.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * 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.core.execution;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.DescribedEnum;
+import org.apache.flink.configuration.description.InlineElement;
+
+import static org.apache.flink.configuration.description.TextElement.text;
+
+/** Describes the backup type in which a checkpoint should be taken. */
+@PublicEvolving
+public enum CheckpointBackupType implements DescribedEnum {
Review Comment:
Thanks for the context @pnowojski . then we need to add the `@Internal` annotation to the `org.apache.flink.runtime.checkpoint.CheckpointType` class. I didn't see the annotation and hence thought it is intended as public.
With that context, I would agree with Piotr's suggestion on renaming the internal class. maybe we can rename `org.apache.flink.runtime.checkpoint.CheckpointType` as `org.apache.flink.runtime.checkpoint.SnapshotTypeImpl` and annotate it with `@Internal`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zentol commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
zentol commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r992210095
##########
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java:
##########
@@ -511,6 +514,47 @@ public CompletableFuture<CompletedCheckpoint> triggerCheckpoint(boolean isPeriod
return triggerCheckpointFromCheckpointThread(checkpointProperties, null, isPeriodic);
}
+ /**
+ * Triggers one new checkpoint with the given checkpointType. The return value is a future. It
+ * completes when the checkpoint triggered finishes or an error occurred.
Review Comment:
```suggestion
* Triggers one new checkpoint with the given checkpointType. The returned future
* completes when the triggered checkpoint finishes or an error occurred.
```
##########
flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/checkpoints/CheckpointTriggerInfo.java:
##########
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.rest.messages.checkpoints;
+
+import org.apache.flink.runtime.rest.messages.ResponseBody;
+import org.apache.flink.runtime.rest.messages.json.SerializedThrowableDeserializer;
+import org.apache.flink.runtime.rest.messages.json.SerializedThrowableSerializer;
+import org.apache.flink.util.SerializedThrowable;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnore;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.annotation.JsonSerialize;
+
+import javax.annotation.Nullable;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+
+/** Represents information about a finished triggered checkpoint. */
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public class CheckpointTriggerInfo implements ResponseBody {
+
+ private static final String FIELD_NAME_CHECKPOINT_ID = "checkpointId";
+
+ private static final String FIELD_NAME_FAILURE_CAUSE = "failureCause";
+
+ @JsonProperty(FIELD_NAME_CHECKPOINT_ID)
+ @Nullable
+ private final Long checkpointId;
+
+ @JsonProperty(FIELD_NAME_FAILURE_CAUSE)
+ @JsonSerialize(using = SerializedThrowableSerializer.class)
+ @JsonDeserialize(using = SerializedThrowableDeserializer.class)
+ @Nullable
+ private final SerializedThrowable failureCause;
+
+ @JsonCreator
+ public CheckpointTriggerInfo(
+ @JsonProperty(FIELD_NAME_CHECKPOINT_ID) @Nullable final Long checkpointId,
+ @JsonProperty(FIELD_NAME_FAILURE_CAUSE)
+ @JsonDeserialize(using = SerializedThrowableDeserializer.class)
+ @Nullable
+ final SerializedThrowable failureCause) {
+ checkArgument(
+ checkpointId != null ^ failureCause != null,
+ "Either checkpointId or failureCause must be set");
+
+ this.checkpointId = checkpointId;
+ this.failureCause = failureCause;
+ }
+
+ @Nullable
+ public Long getCheckpointId() {
Review Comment:
```suggestion
@Nullable
@JsonIgnore
public Long getCheckpointId() {
```
##########
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java:
##########
@@ -511,6 +514,47 @@ public CompletableFuture<CompletedCheckpoint> triggerCheckpoint(boolean isPeriod
return triggerCheckpointFromCheckpointThread(checkpointProperties, null, isPeriodic);
}
+ /**
+ * Triggers one new checkpoint with the given checkpointType. The return value is a future. It
+ * completes when the checkpoint triggered finishes or an error occurred.
+ *
+ * @param checkpointType specifies the back up type of the checkpoint to trigger.
+ * @return a future to the completed checkpoint.
+ */
+ public CompletableFuture<CompletedCheckpoint> triggerCheckpoint(CheckpointType checkpointType) {
+
+ if (checkpointType == null) {
+ throw new IllegalArgumentException("checkpointType cannot be null");
+ }
+
+ final SnapshotType snapshotType;
+ switch (checkpointType) {
+ case CONFIGURED:
+ snapshotType = checkpointProperties.getCheckpointType();
+ break;
+ case FULL:
+ snapshotType = FULL_CHECKPOINT;
+ break;
+ case INCREMENTAL:
+ snapshotType = CHECKPOINT;
+ break;
+ default:
+ throw new IllegalArgumentException("unknown checkpointType: " + checkpointType);
+ }
+
+ CheckpointProperties properties =
Review Comment:
```suggestion
final CheckpointProperties properties =
```
##########
flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/checkpoints/CheckpointTriggerHandlers.java:
##########
@@ -0,0 +1,261 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.rest.handler.job.checkpoints;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.core.execution.CheckpointType;
+import org.apache.flink.runtime.dispatcher.UnknownOperationKeyException;
+import org.apache.flink.runtime.rest.handler.AbstractRestHandler;
+import org.apache.flink.runtime.rest.handler.HandlerRequest;
+import org.apache.flink.runtime.rest.handler.RestHandlerException;
+import org.apache.flink.runtime.rest.handler.async.AsynchronousOperationResult;
+import org.apache.flink.runtime.rest.handler.async.TriggerResponse;
+import org.apache.flink.runtime.rest.handler.job.AsynchronousJobOperationKey;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.JobIDPathParameter;
+import org.apache.flink.runtime.rest.messages.TriggerId;
+import org.apache.flink.runtime.rest.messages.TriggerIdPathParameter;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerHeaders;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerInfo;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerMessageParameters;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerRequestBody;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerStatusHeaders;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerStatusMessageParameters;
+import org.apache.flink.runtime.rpc.RpcUtils;
+import org.apache.flink.runtime.webmonitor.RestfulGateway;
+import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.SerializedThrowable;
+
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+import javax.annotation.Nonnull;
+
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+
+/**
+ * HTTP handlers for asynchronous triggering of checkpoints.
+ *
+ * <p>Drawing checkpoints is a potentially long-running operation. To avoid blocking HTTP
+ * connections, checkpoints must be drawn in two steps. First, an HTTP request is issued to trigger
+ * the checkpoint asynchronously. The request will be assigned a {@link TriggerId}, which is
+ * returned in the response body. Next, the returned {@link TriggerId} should be used to poll the
+ * status of the checkpoint until it is finished.
+ *
+ * <p>A checkpoint is triggered by sending an HTTP {@code POST} request to {@code
+ * /jobs/:jobid/checkpoints}. The HTTP request may contain a JSON body to specify a customized
+ * {@link TriggerId} and a {@link CheckpointType}, e.g.,
+ *
+ * <pre>
+ * { "triggerId": "7d273f5a62eb4730b9dea8e833733c1e", "checkpointType": "FULL" }
+ * </pre>
+ *
+ * <p>If the body is omitted, or the field {@code checkpointType} is {@code null}, the default
+ * checkpointType as specified by {@link CheckpointType#CONFIGURED} will be used. As written above,
+ * the response will contain a request id, e.g.,
+ *
+ * <pre>
+ * { "request-id": "7d273f5a62eb4730b9dea8e833733c1e" }
+ * </pre>
+ *
+ * <p>To poll for the status of an ongoing checkpoint, an HTTP {@code GET} request is issued to
+ * {@code /jobs/:jobid/checkpoints/status/:checkpointtriggerid}. If the specified checkpoint is
+ * still ongoing, the response will be
+ *
+ * <pre>
+ * {
+ * "status": {
+ * "id": "IN_PROGRESS"
+ * }
+ * }
+ * </pre>
+ *
+ * <p>If the specified checkpoints has completed, the status id will transition to {@code
+ * COMPLETED}, and the response will additionally contain information about the savepoint, such as
+ * the location:
+ *
+ * <pre>
+ * {
+ * "status": {
+ * "id": "COMPLETED"
+ * },
+ * "operation": {
+ * "checkpointId": "123"
+ * }
+ * }
+ * </pre>
+ */
+public class CheckpointTriggerHandlers {
+
+ /** handler for the checkpoint trigger operation. */
+ public static class CheckpointTriggerHandler
+ extends AbstractRestHandler<
+ RestfulGateway,
+ CheckpointTriggerRequestBody,
+ TriggerResponse,
+ CheckpointTriggerMessageParameters> {
+
+ public CheckpointTriggerHandler(
+ final GatewayRetriever<? extends RestfulGateway> leaderRetriever,
+ final Time timeout,
+ final Map<String, String> responseHeaders) {
+ super(
+ leaderRetriever,
+ timeout,
+ responseHeaders,
+ CheckpointTriggerHeaders.getInstance());
+ }
+
+ protected AsynchronousJobOperationKey createOperationKey(
Review Comment:
```suggestion
private static AsynchronousJobOperationKey createOperationKey(
```
##########
flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/checkpoints/CheckpointTriggerStatusHeaders.java:
##########
@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.rest.messages.checkpoints;
+
+import org.apache.flink.runtime.rest.HttpMethodWrapper;
+import org.apache.flink.runtime.rest.handler.async.AsynchronousOperationStatusMessageHeaders;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.JobIDPathParameter;
+import org.apache.flink.runtime.rest.messages.TriggerIdPathParameter;
+
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+/** These headers define the protocol for triggering a checkpoint. */
Review Comment:
```suggestion
/** These headers define the protocol for querying the status of a checkpoint operation. */
```
##########
flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/savepoints/SavepointInfo.java:
##########
@@ -72,6 +73,7 @@ public String getLocation() {
}
@Nullable
+ @JsonIgnore
Review Comment:
also apply this to `getLocation`
##########
flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/checkpoints/CheckpointTriggerHandlers.java:
##########
@@ -0,0 +1,261 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.rest.handler.job.checkpoints;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.core.execution.CheckpointType;
+import org.apache.flink.runtime.dispatcher.UnknownOperationKeyException;
+import org.apache.flink.runtime.rest.handler.AbstractRestHandler;
+import org.apache.flink.runtime.rest.handler.HandlerRequest;
+import org.apache.flink.runtime.rest.handler.RestHandlerException;
+import org.apache.flink.runtime.rest.handler.async.AsynchronousOperationResult;
+import org.apache.flink.runtime.rest.handler.async.TriggerResponse;
+import org.apache.flink.runtime.rest.handler.job.AsynchronousJobOperationKey;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.JobIDPathParameter;
+import org.apache.flink.runtime.rest.messages.TriggerId;
+import org.apache.flink.runtime.rest.messages.TriggerIdPathParameter;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerHeaders;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerInfo;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerMessageParameters;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerRequestBody;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerStatusHeaders;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerStatusMessageParameters;
+import org.apache.flink.runtime.rpc.RpcUtils;
+import org.apache.flink.runtime.webmonitor.RestfulGateway;
+import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.SerializedThrowable;
+
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+import javax.annotation.Nonnull;
+
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+
+/**
+ * HTTP handlers for asynchronous triggering of checkpoints.
+ *
+ * <p>Drawing checkpoints is a potentially long-running operation. To avoid blocking HTTP
+ * connections, checkpoints must be drawn in two steps. First, an HTTP request is issued to trigger
+ * the checkpoint asynchronously. The request will be assigned a {@link TriggerId}, which is
+ * returned in the response body. Next, the returned {@link TriggerId} should be used to poll the
+ * status of the checkpoint until it is finished.
+ *
+ * <p>A checkpoint is triggered by sending an HTTP {@code POST} request to {@code
+ * /jobs/:jobid/checkpoints}. The HTTP request may contain a JSON body to specify a customized
+ * {@link TriggerId} and a {@link CheckpointType}, e.g.,
+ *
+ * <pre>
+ * { "triggerId": "7d273f5a62eb4730b9dea8e833733c1e", "checkpointType": "FULL" }
+ * </pre>
+ *
+ * <p>If the body is omitted, or the field {@code checkpointType} is {@code null}, the default
+ * checkpointType as specified by {@link CheckpointType#CONFIGURED} will be used. As written above,
+ * the response will contain a request id, e.g.,
+ *
+ * <pre>
+ * { "request-id": "7d273f5a62eb4730b9dea8e833733c1e" }
+ * </pre>
+ *
+ * <p>To poll for the status of an ongoing checkpoint, an HTTP {@code GET} request is issued to
+ * {@code /jobs/:jobid/checkpoints/status/:checkpointtriggerid}. If the specified checkpoint is
+ * still ongoing, the response will be
+ *
+ * <pre>
+ * {
+ * "status": {
+ * "id": "IN_PROGRESS"
+ * }
+ * }
+ * </pre>
+ *
+ * <p>If the specified checkpoints has completed, the status id will transition to {@code
+ * COMPLETED}, and the response will additionally contain information about the savepoint, such as
+ * the location:
+ *
+ * <pre>
+ * {
+ * "status": {
+ * "id": "COMPLETED"
+ * },
+ * "operation": {
+ * "checkpointId": "123"
+ * }
+ * }
+ * </pre>
+ */
+public class CheckpointTriggerHandlers {
+
+ /** handler for the checkpoint trigger operation. */
Review Comment:
```suggestion
/** Handler for the checkpoint trigger operation. */
```
##########
flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/checkpoints/CheckpointTriggerHandlersTest.java:
##########
@@ -0,0 +1,349 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.rest.handler.job.checkpoints;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.core.execution.CheckpointType;
+import org.apache.flink.runtime.dispatcher.UnknownOperationKeyException;
+import org.apache.flink.runtime.messages.Acknowledge;
+import org.apache.flink.runtime.rest.RestMatchers;
+import org.apache.flink.runtime.rest.handler.HandlerRequest;
+import org.apache.flink.runtime.rest.handler.HandlerRequestException;
+import org.apache.flink.runtime.rest.handler.RestHandlerException;
+import org.apache.flink.runtime.rest.handler.async.AsynchronousOperationResult;
+import org.apache.flink.runtime.rest.handler.async.OperationResult;
+import org.apache.flink.runtime.rest.handler.job.AsynchronousJobOperationKey;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.JobIDPathParameter;
+import org.apache.flink.runtime.rest.messages.TriggerId;
+import org.apache.flink.runtime.rest.messages.TriggerIdPathParameter;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerInfo;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerMessageParameters;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerRequestBody;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerStatusMessageParameters;
+import org.apache.flink.runtime.rest.messages.queue.QueueStatus;
+import org.apache.flink.runtime.webmonitor.RestfulGateway;
+import org.apache.flink.runtime.webmonitor.TestingRestfulGateway;
+import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
+import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.concurrent.FutureUtils;
+
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/** Test for {@link CheckpointTriggerHandlers}. */
+public class CheckpointTriggerHandlersTest extends TestLogger {
+
+ private static final Time TIMEOUT = Time.seconds(10);
+
+ private static final JobID JOB_ID = new JobID();
+
+ private static final Long COMPLETED_CHECKPOINT_ID = 123456L;
+
+ private static CheckpointTriggerHandlers.CheckpointTriggerHandler checkpointTriggerHandler;
+
+ private static CheckpointTriggerHandlers.CheckpointTriggerStatusHandler
+ checkpointTriggerStatusHandler;
+
+ @BeforeAll
+ static void setUp() throws Exception {
+ GatewayRetriever<RestfulGateway> leaderRetriever =
+ () -> CompletableFuture.completedFuture(null);
+
+ checkpointTriggerHandler =
+ new CheckpointTriggerHandlers.CheckpointTriggerHandler(
+ leaderRetriever, TIMEOUT, Collections.emptyMap());
+
+ checkpointTriggerStatusHandler =
+ new CheckpointTriggerHandlers.CheckpointTriggerStatusHandler(
+ leaderRetriever, TIMEOUT, Collections.emptyMap());
+ }
+
+ @Test
+ public void testCheckpointTriggerCompletedSuccessfully() throws Exception {
Review Comment:
```suggestion
void testCheckpointTriggerCompletedSuccessfully() throws Exception {
```
##########
flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/checkpoints/CheckpointTriggerHandlersTest.java:
##########
@@ -0,0 +1,349 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.rest.handler.job.checkpoints;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.core.execution.CheckpointType;
+import org.apache.flink.runtime.dispatcher.UnknownOperationKeyException;
+import org.apache.flink.runtime.messages.Acknowledge;
+import org.apache.flink.runtime.rest.RestMatchers;
+import org.apache.flink.runtime.rest.handler.HandlerRequest;
+import org.apache.flink.runtime.rest.handler.HandlerRequestException;
+import org.apache.flink.runtime.rest.handler.RestHandlerException;
+import org.apache.flink.runtime.rest.handler.async.AsynchronousOperationResult;
+import org.apache.flink.runtime.rest.handler.async.OperationResult;
+import org.apache.flink.runtime.rest.handler.job.AsynchronousJobOperationKey;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.JobIDPathParameter;
+import org.apache.flink.runtime.rest.messages.TriggerId;
+import org.apache.flink.runtime.rest.messages.TriggerIdPathParameter;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerInfo;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerMessageParameters;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerRequestBody;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerStatusMessageParameters;
+import org.apache.flink.runtime.rest.messages.queue.QueueStatus;
+import org.apache.flink.runtime.webmonitor.RestfulGateway;
+import org.apache.flink.runtime.webmonitor.TestingRestfulGateway;
+import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
+import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.concurrent.FutureUtils;
+
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/** Test for {@link CheckpointTriggerHandlers}. */
+public class CheckpointTriggerHandlersTest extends TestLogger {
Review Comment:
```suggestion
class CheckpointTriggerHandlersTest extends TestLogger {
```
##########
flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/checkpoints/CheckpointTriggerHandlers.java:
##########
@@ -0,0 +1,261 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.rest.handler.job.checkpoints;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.core.execution.CheckpointType;
+import org.apache.flink.runtime.dispatcher.UnknownOperationKeyException;
+import org.apache.flink.runtime.rest.handler.AbstractRestHandler;
+import org.apache.flink.runtime.rest.handler.HandlerRequest;
+import org.apache.flink.runtime.rest.handler.RestHandlerException;
+import org.apache.flink.runtime.rest.handler.async.AsynchronousOperationResult;
+import org.apache.flink.runtime.rest.handler.async.TriggerResponse;
+import org.apache.flink.runtime.rest.handler.job.AsynchronousJobOperationKey;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.JobIDPathParameter;
+import org.apache.flink.runtime.rest.messages.TriggerId;
+import org.apache.flink.runtime.rest.messages.TriggerIdPathParameter;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerHeaders;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerInfo;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerMessageParameters;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerRequestBody;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerStatusHeaders;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerStatusMessageParameters;
+import org.apache.flink.runtime.rpc.RpcUtils;
+import org.apache.flink.runtime.webmonitor.RestfulGateway;
+import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.SerializedThrowable;
+
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+import javax.annotation.Nonnull;
+
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+
+/**
+ * HTTP handlers for asynchronous triggering of checkpoints.
+ *
+ * <p>Drawing checkpoints is a potentially long-running operation. To avoid blocking HTTP
+ * connections, checkpoints must be drawn in two steps. First, an HTTP request is issued to trigger
+ * the checkpoint asynchronously. The request will be assigned a {@link TriggerId}, which is
+ * returned in the response body. Next, the returned {@link TriggerId} should be used to poll the
+ * status of the checkpoint until it is finished.
+ *
+ * <p>A checkpoint is triggered by sending an HTTP {@code POST} request to {@code
+ * /jobs/:jobid/checkpoints}. The HTTP request may contain a JSON body to specify a customized
+ * {@link TriggerId} and a {@link CheckpointType}, e.g.,
+ *
+ * <pre>
+ * { "triggerId": "7d273f5a62eb4730b9dea8e833733c1e", "checkpointType": "FULL" }
+ * </pre>
+ *
+ * <p>If the body is omitted, or the field {@code checkpointType} is {@code null}, the default
+ * checkpointType as specified by {@link CheckpointType#CONFIGURED} will be used. As written above,
+ * the response will contain a request id, e.g.,
+ *
+ * <pre>
+ * { "request-id": "7d273f5a62eb4730b9dea8e833733c1e" }
+ * </pre>
+ *
+ * <p>To poll for the status of an ongoing checkpoint, an HTTP {@code GET} request is issued to
+ * {@code /jobs/:jobid/checkpoints/status/:checkpointtriggerid}. If the specified checkpoint is
+ * still ongoing, the response will be
+ *
+ * <pre>
+ * {
+ * "status": {
+ * "id": "IN_PROGRESS"
+ * }
+ * }
+ * </pre>
+ *
+ * <p>If the specified checkpoints has completed, the status id will transition to {@code
+ * COMPLETED}, and the response will additionally contain information about the savepoint, such as
+ * the location:
+ *
+ * <pre>
+ * {
+ * "status": {
+ * "id": "COMPLETED"
+ * },
+ * "operation": {
+ * "checkpointId": "123"
+ * }
+ * }
+ * </pre>
+ */
+public class CheckpointTriggerHandlers {
+
+ /** handler for the checkpoint trigger operation. */
+ public static class CheckpointTriggerHandler
+ extends AbstractRestHandler<
+ RestfulGateway,
+ CheckpointTriggerRequestBody,
+ TriggerResponse,
+ CheckpointTriggerMessageParameters> {
+
+ public CheckpointTriggerHandler(
+ final GatewayRetriever<? extends RestfulGateway> leaderRetriever,
+ final Time timeout,
+ final Map<String, String> responseHeaders) {
+ super(
+ leaderRetriever,
+ timeout,
+ responseHeaders,
+ CheckpointTriggerHeaders.getInstance());
+ }
+
+ protected AsynchronousJobOperationKey createOperationKey(
+ final HandlerRequest<CheckpointTriggerRequestBody> request) {
+ final JobID jobId = request.getPathParameter(JobIDPathParameter.class);
+ return AsynchronousJobOperationKey.of(
+ request.getRequestBody().getTriggerId().orElseGet(TriggerId::new), jobId);
+ }
+
+ @Override
+ protected CompletableFuture<TriggerResponse> handleRequest(
+ @Nonnull HandlerRequest<CheckpointTriggerRequestBody> request,
+ @Nonnull RestfulGateway gateway)
+ throws RestHandlerException {
+ final AsynchronousJobOperationKey operationKey = createOperationKey(request);
+
+ return gateway.triggerCheckpoint(
+ operationKey,
+ request.getRequestBody().getCheckpointType(),
+ RpcUtils.INF_TIMEOUT)
+ .handle(
+ (acknowledge, throwable) -> {
+ if (throwable == null) {
+ return new TriggerResponse(operationKey.getTriggerId());
+ } else {
+ throw new CompletionException(
+ createInternalServerError(
+ throwable, operationKey, "triggering"));
+ }
+ });
+ }
+ }
+
+ /** HTTP handler to query for the status of the checkpoint trigger. */
+ public static class CheckpointTriggerStatusHandler
+ extends AbstractRestHandler<
+ RestfulGateway,
+ EmptyRequestBody,
+ AsynchronousOperationResult<CheckpointTriggerInfo>,
+ CheckpointTriggerStatusMessageParameters> {
+
+ public CheckpointTriggerStatusHandler(
+ final GatewayRetriever<? extends RestfulGateway> leaderRetriever,
+ final Time timeout,
+ final Map<String, String> responseHeaders) {
+ super(
+ leaderRetriever,
+ timeout,
+ responseHeaders,
+ CheckpointTriggerStatusHeaders.getInstance());
+ }
+
+ @Override
+ public CompletableFuture<AsynchronousOperationResult<CheckpointTriggerInfo>> handleRequest(
+ @Nonnull HandlerRequest<EmptyRequestBody> request, @Nonnull RestfulGateway gateway)
+ throws RestHandlerException {
+
+ final AsynchronousJobOperationKey key = getOperationKey(request);
+
+ return gateway.getTriggeredCheckpointStatus(key)
+ .handle(
+ (operationResult, throwable) -> {
+ if (throwable == null) {
+ switch (operationResult.getStatus()) {
+ case SUCCESS:
+ return AsynchronousOperationResult.completed(
+ operationResultResponse(
+ operationResult.getResult()));
+ case FAILURE:
+ return AsynchronousOperationResult.completed(
+ exceptionalOperationResultResponse(
+ operationResult.getThrowable()));
+ case IN_PROGRESS:
+ return AsynchronousOperationResult.inProgress();
+ default:
+ throw new IllegalStateException(
+ "No handler for operation status "
+ + operationResult.getStatus()
+ + ", encountered for key "
+ + key);
+ }
+ } else {
+ throw new CompletionException(
+ maybeCreateNotFoundError(throwable, key)
+ .orElseGet(
+ () ->
+ createInternalServerError(
+ throwable,
+ key,
+ "retrieving status of")));
+ }
+ });
+ }
+
+ private static Optional<RestHandlerException> maybeCreateNotFoundError(
+ Throwable throwable, AsynchronousJobOperationKey key) {
+ if (ExceptionUtils.findThrowable(throwable, UnknownOperationKeyException.class)
+ .isPresent()) {
+ return Optional.of(
+ new RestHandlerException(
+ String.format(
+ "There is no checkpoint operation with triggerId=%s for job %s.",
+ key.getTriggerId(), key.getJobId()),
+ HttpResponseStatus.NOT_FOUND));
+ }
+ return Optional.empty();
+ }
+
+ protected AsynchronousJobOperationKey getOperationKey(
+ HandlerRequest<EmptyRequestBody> request) {
+ final TriggerId triggerId = request.getPathParameter(TriggerIdPathParameter.class);
+ final JobID jobId = request.getPathParameter(JobIDPathParameter.class);
+ return AsynchronousJobOperationKey.of(triggerId, jobId);
+ }
+
+ protected CheckpointTriggerInfo exceptionalOperationResultResponse(
+ final Throwable throwable) {
+ return new CheckpointTriggerInfo(null, new SerializedThrowable(throwable));
+ }
Review Comment:
```suggestion
private static CheckpointTriggerInfo exceptionalOperationResultResponse(
final Throwable throwable) {
return new CheckpointTriggerInfo(null, new SerializedThrowable(throwable));
}
```
##########
flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/checkpoints/CheckpointTriggerInfo.java:
##########
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.rest.messages.checkpoints;
+
+import org.apache.flink.runtime.rest.messages.ResponseBody;
+import org.apache.flink.runtime.rest.messages.json.SerializedThrowableDeserializer;
+import org.apache.flink.runtime.rest.messages.json.SerializedThrowableSerializer;
+import org.apache.flink.util.SerializedThrowable;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnore;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.annotation.JsonSerialize;
+
+import javax.annotation.Nullable;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+
+/** Represents information about a finished triggered checkpoint. */
Review Comment:
```suggestion
/** Represents information about a triggered checkpoint. */
```
##########
flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/checkpoints/CheckpointTriggerHandlersTest.java:
##########
@@ -0,0 +1,349 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.rest.handler.job.checkpoints;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.core.execution.CheckpointType;
+import org.apache.flink.runtime.dispatcher.UnknownOperationKeyException;
+import org.apache.flink.runtime.messages.Acknowledge;
+import org.apache.flink.runtime.rest.RestMatchers;
+import org.apache.flink.runtime.rest.handler.HandlerRequest;
+import org.apache.flink.runtime.rest.handler.HandlerRequestException;
+import org.apache.flink.runtime.rest.handler.RestHandlerException;
+import org.apache.flink.runtime.rest.handler.async.AsynchronousOperationResult;
+import org.apache.flink.runtime.rest.handler.async.OperationResult;
+import org.apache.flink.runtime.rest.handler.job.AsynchronousJobOperationKey;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.JobIDPathParameter;
+import org.apache.flink.runtime.rest.messages.TriggerId;
+import org.apache.flink.runtime.rest.messages.TriggerIdPathParameter;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerInfo;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerMessageParameters;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerRequestBody;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerStatusMessageParameters;
+import org.apache.flink.runtime.rest.messages.queue.QueueStatus;
+import org.apache.flink.runtime.webmonitor.RestfulGateway;
+import org.apache.flink.runtime.webmonitor.TestingRestfulGateway;
+import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
+import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.concurrent.FutureUtils;
+
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/** Test for {@link CheckpointTriggerHandlers}. */
+public class CheckpointTriggerHandlersTest extends TestLogger {
+
+ private static final Time TIMEOUT = Time.seconds(10);
+
+ private static final JobID JOB_ID = new JobID();
+
+ private static final Long COMPLETED_CHECKPOINT_ID = 123456L;
+
+ private static CheckpointTriggerHandlers.CheckpointTriggerHandler checkpointTriggerHandler;
+
+ private static CheckpointTriggerHandlers.CheckpointTriggerStatusHandler
+ checkpointTriggerStatusHandler;
+
+ @BeforeAll
+ static void setUp() throws Exception {
+ GatewayRetriever<RestfulGateway> leaderRetriever =
+ () -> CompletableFuture.completedFuture(null);
+
+ checkpointTriggerHandler =
+ new CheckpointTriggerHandlers.CheckpointTriggerHandler(
+ leaderRetriever, TIMEOUT, Collections.emptyMap());
+
+ checkpointTriggerStatusHandler =
+ new CheckpointTriggerHandlers.CheckpointTriggerStatusHandler(
+ leaderRetriever, TIMEOUT, Collections.emptyMap());
+ }
+
+ @Test
+ public void testCheckpointTriggerCompletedSuccessfully() throws Exception {
+ final OperationResult<Long> successfulResult =
+ OperationResult.success(COMPLETED_CHECKPOINT_ID);
+ final CompletableFuture<CheckpointType> checkpointPropertiesFuture =
+ new CompletableFuture<>();
+
+ final AtomicReference<AsynchronousJobOperationKey> keyReference = new AtomicReference<>();
+ final TestingRestfulGateway testingRestfulGateway =
+ new TestingRestfulGateway.Builder()
+ .setTriggerCheckpointFunction(
+ (AsynchronousJobOperationKey key,
+ CheckpointType checkpointType) -> {
+ keyReference.set(key);
+ checkpointPropertiesFuture.complete(checkpointType);
+ return CompletableFuture.completedFuture(Acknowledge.get());
+ })
+ .setGetCheckpointStatusFunction(
+ (AsynchronousJobOperationKey operationKey) -> {
+ if (operationKey.equals(keyReference.get())) {
+ return CompletableFuture.completedFuture(successfulResult);
+ }
+ throw new RuntimeException(
+ "Expected operation key "
+ + keyReference.get()
+ + ", but received "
+ + operationKey);
+ })
+ .build();
+
+ final CheckpointType checkpointType = CheckpointType.FULL;
+
+ final TriggerId triggerId =
+ checkpointTriggerHandler
+ .handleRequest(
+ triggerCheckpointRequest(checkpointType, null),
+ testingRestfulGateway)
+ .get()
+ .getTriggerId();
+
+ final AsynchronousOperationResult<CheckpointTriggerInfo> checkpointTriggerResponseBody =
+ checkpointTriggerStatusHandler
+ .handleRequest(
+ checkpointTriggerStatusRequest(triggerId), testingRestfulGateway)
+ .get();
+
+ assertEquals(checkpointTriggerResponseBody.queueStatus().getId(), QueueStatus.Id.COMPLETED);
+ assertNotNull(checkpointTriggerResponseBody.resource());
+ assertEquals(
+ checkpointTriggerResponseBody.resource().getCheckpointId(),
+ COMPLETED_CHECKPOINT_ID);
+ assertEquals(checkpointPropertiesFuture.get(), CheckpointType.FULL);
+ }
+
+ @Test
+ public void testTriggerCheckpointNoCheckpointType() throws Exception {
+ final OperationResult<Long> successfulResult =
+ OperationResult.success(COMPLETED_CHECKPOINT_ID);
+ final CompletableFuture<CheckpointType> checkpointTypeFuture = new CompletableFuture<>();
+
+ final AtomicReference<AsynchronousJobOperationKey> keyReference = new AtomicReference<>();
+ final TestingRestfulGateway testingRestfulGateway =
+ new TestingRestfulGateway.Builder()
+ .setTriggerCheckpointFunction(
+ (AsynchronousJobOperationKey key,
+ CheckpointType checkpointType) -> {
+ keyReference.set(key);
+ checkpointTypeFuture.complete(checkpointType);
+ return CompletableFuture.completedFuture(Acknowledge.get());
+ })
+ .setGetCheckpointStatusFunction(
+ (AsynchronousJobOperationKey operationKey) -> {
+ if (operationKey.equals(keyReference.get())) {
+ return CompletableFuture.completedFuture(successfulResult);
+ }
+ throw new RuntimeException(
+ "Expected operation key "
+ + keyReference.get()
+ + ", but received "
+ + operationKey);
+ })
+ .build();
+
+ final TriggerId triggerId =
+ checkpointTriggerHandler
+ .handleRequest(triggerCheckpointRequest(null, null), testingRestfulGateway)
+ .get()
+ .getTriggerId();
+
+ AsynchronousOperationResult<CheckpointTriggerInfo> checkpointTriggerResponseBody;
+ checkpointTriggerResponseBody =
+ checkpointTriggerStatusHandler
+ .handleRequest(
+ checkpointTriggerStatusRequest(triggerId), testingRestfulGateway)
+ .get();
+
+ assertEquals(checkpointTriggerResponseBody.queueStatus().getId(), QueueStatus.Id.COMPLETED);
+ assertNotNull(checkpointTriggerResponseBody.resource());
+ assertEquals(
+ checkpointTriggerResponseBody.resource().getCheckpointId(),
+ COMPLETED_CHECKPOINT_ID);
+ assertEquals(checkpointTypeFuture.get(), CheckpointType.DEFAULT);
+ }
+
+ @Test
+ public void testCheckpointCompletedWithException() throws Exception {
Review Comment:
```suggestion
void testCheckpointCompletedWithException() throws Exception {
```
##########
flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/checkpoints/CheckpointTriggerHandlersTest.java:
##########
@@ -0,0 +1,349 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.rest.handler.job.checkpoints;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.core.execution.CheckpointType;
+import org.apache.flink.runtime.dispatcher.UnknownOperationKeyException;
+import org.apache.flink.runtime.messages.Acknowledge;
+import org.apache.flink.runtime.rest.RestMatchers;
+import org.apache.flink.runtime.rest.handler.HandlerRequest;
+import org.apache.flink.runtime.rest.handler.HandlerRequestException;
+import org.apache.flink.runtime.rest.handler.RestHandlerException;
+import org.apache.flink.runtime.rest.handler.async.AsynchronousOperationResult;
+import org.apache.flink.runtime.rest.handler.async.OperationResult;
+import org.apache.flink.runtime.rest.handler.job.AsynchronousJobOperationKey;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.JobIDPathParameter;
+import org.apache.flink.runtime.rest.messages.TriggerId;
+import org.apache.flink.runtime.rest.messages.TriggerIdPathParameter;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerInfo;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerMessageParameters;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerRequestBody;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerStatusMessageParameters;
+import org.apache.flink.runtime.rest.messages.queue.QueueStatus;
+import org.apache.flink.runtime.webmonitor.RestfulGateway;
+import org.apache.flink.runtime.webmonitor.TestingRestfulGateway;
+import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
+import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.concurrent.FutureUtils;
+
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/** Test for {@link CheckpointTriggerHandlers}. */
+public class CheckpointTriggerHandlersTest extends TestLogger {
+
+ private static final Time TIMEOUT = Time.seconds(10);
+
+ private static final JobID JOB_ID = new JobID();
+
+ private static final Long COMPLETED_CHECKPOINT_ID = 123456L;
+
+ private static CheckpointTriggerHandlers.CheckpointTriggerHandler checkpointTriggerHandler;
+
+ private static CheckpointTriggerHandlers.CheckpointTriggerStatusHandler
+ checkpointTriggerStatusHandler;
+
+ @BeforeAll
+ static void setUp() throws Exception {
+ GatewayRetriever<RestfulGateway> leaderRetriever =
+ () -> CompletableFuture.completedFuture(null);
+
+ checkpointTriggerHandler =
+ new CheckpointTriggerHandlers.CheckpointTriggerHandler(
+ leaderRetriever, TIMEOUT, Collections.emptyMap());
+
+ checkpointTriggerStatusHandler =
+ new CheckpointTriggerHandlers.CheckpointTriggerStatusHandler(
+ leaderRetriever, TIMEOUT, Collections.emptyMap());
+ }
+
+ @Test
+ public void testCheckpointTriggerCompletedSuccessfully() throws Exception {
+ final OperationResult<Long> successfulResult =
+ OperationResult.success(COMPLETED_CHECKPOINT_ID);
+ final CompletableFuture<CheckpointType> checkpointPropertiesFuture =
+ new CompletableFuture<>();
+
+ final AtomicReference<AsynchronousJobOperationKey> keyReference = new AtomicReference<>();
+ final TestingRestfulGateway testingRestfulGateway =
+ new TestingRestfulGateway.Builder()
+ .setTriggerCheckpointFunction(
+ (AsynchronousJobOperationKey key,
+ CheckpointType checkpointType) -> {
+ keyReference.set(key);
+ checkpointPropertiesFuture.complete(checkpointType);
+ return CompletableFuture.completedFuture(Acknowledge.get());
+ })
+ .setGetCheckpointStatusFunction(
+ (AsynchronousJobOperationKey operationKey) -> {
+ if (operationKey.equals(keyReference.get())) {
+ return CompletableFuture.completedFuture(successfulResult);
+ }
+ throw new RuntimeException(
+ "Expected operation key "
+ + keyReference.get()
+ + ", but received "
+ + operationKey);
+ })
+ .build();
+
+ final CheckpointType checkpointType = CheckpointType.FULL;
+
+ final TriggerId triggerId =
+ checkpointTriggerHandler
+ .handleRequest(
+ triggerCheckpointRequest(checkpointType, null),
+ testingRestfulGateway)
+ .get()
+ .getTriggerId();
+
+ final AsynchronousOperationResult<CheckpointTriggerInfo> checkpointTriggerResponseBody =
+ checkpointTriggerStatusHandler
+ .handleRequest(
+ checkpointTriggerStatusRequest(triggerId), testingRestfulGateway)
+ .get();
+
+ assertEquals(checkpointTriggerResponseBody.queueStatus().getId(), QueueStatus.Id.COMPLETED);
+ assertNotNull(checkpointTriggerResponseBody.resource());
+ assertEquals(
+ checkpointTriggerResponseBody.resource().getCheckpointId(),
+ COMPLETED_CHECKPOINT_ID);
+ assertEquals(checkpointPropertiesFuture.get(), CheckpointType.FULL);
+ }
+
+ @Test
+ public void testTriggerCheckpointNoCheckpointType() throws Exception {
Review Comment:
```suggestion
void testTriggerCheckpointNoCheckpointType() throws Exception {
```
##########
flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/checkpoints/CheckpointTriggerHandlers.java:
##########
@@ -0,0 +1,261 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.rest.handler.job.checkpoints;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.core.execution.CheckpointType;
+import org.apache.flink.runtime.dispatcher.UnknownOperationKeyException;
+import org.apache.flink.runtime.rest.handler.AbstractRestHandler;
+import org.apache.flink.runtime.rest.handler.HandlerRequest;
+import org.apache.flink.runtime.rest.handler.RestHandlerException;
+import org.apache.flink.runtime.rest.handler.async.AsynchronousOperationResult;
+import org.apache.flink.runtime.rest.handler.async.TriggerResponse;
+import org.apache.flink.runtime.rest.handler.job.AsynchronousJobOperationKey;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.JobIDPathParameter;
+import org.apache.flink.runtime.rest.messages.TriggerId;
+import org.apache.flink.runtime.rest.messages.TriggerIdPathParameter;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerHeaders;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerInfo;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerMessageParameters;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerRequestBody;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerStatusHeaders;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerStatusMessageParameters;
+import org.apache.flink.runtime.rpc.RpcUtils;
+import org.apache.flink.runtime.webmonitor.RestfulGateway;
+import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.SerializedThrowable;
+
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+import javax.annotation.Nonnull;
+
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+
+/**
+ * HTTP handlers for asynchronous triggering of checkpoints.
+ *
+ * <p>Drawing checkpoints is a potentially long-running operation. To avoid blocking HTTP
+ * connections, checkpoints must be drawn in two steps. First, an HTTP request is issued to trigger
+ * the checkpoint asynchronously. The request will be assigned a {@link TriggerId}, which is
+ * returned in the response body. Next, the returned {@link TriggerId} should be used to poll the
+ * status of the checkpoint until it is finished.
+ *
+ * <p>A checkpoint is triggered by sending an HTTP {@code POST} request to {@code
+ * /jobs/:jobid/checkpoints}. The HTTP request may contain a JSON body to specify a customized
+ * {@link TriggerId} and a {@link CheckpointType}, e.g.,
+ *
+ * <pre>
+ * { "triggerId": "7d273f5a62eb4730b9dea8e833733c1e", "checkpointType": "FULL" }
+ * </pre>
+ *
+ * <p>If the body is omitted, or the field {@code checkpointType} is {@code null}, the default
+ * checkpointType as specified by {@link CheckpointType#CONFIGURED} will be used. As written above,
+ * the response will contain a request id, e.g.,
+ *
+ * <pre>
+ * { "request-id": "7d273f5a62eb4730b9dea8e833733c1e" }
+ * </pre>
+ *
+ * <p>To poll for the status of an ongoing checkpoint, an HTTP {@code GET} request is issued to
+ * {@code /jobs/:jobid/checkpoints/status/:checkpointtriggerid}. If the specified checkpoint is
+ * still ongoing, the response will be
+ *
+ * <pre>
+ * {
+ * "status": {
+ * "id": "IN_PROGRESS"
+ * }
+ * }
+ * </pre>
+ *
+ * <p>If the specified checkpoints has completed, the status id will transition to {@code
+ * COMPLETED}, and the response will additionally contain information about the savepoint, such as
+ * the location:
+ *
+ * <pre>
+ * {
+ * "status": {
+ * "id": "COMPLETED"
+ * },
+ * "operation": {
+ * "checkpointId": "123"
+ * }
+ * }
+ * </pre>
+ */
+public class CheckpointTriggerHandlers {
+
+ /** handler for the checkpoint trigger operation. */
+ public static class CheckpointTriggerHandler
+ extends AbstractRestHandler<
+ RestfulGateway,
+ CheckpointTriggerRequestBody,
+ TriggerResponse,
+ CheckpointTriggerMessageParameters> {
+
+ public CheckpointTriggerHandler(
+ final GatewayRetriever<? extends RestfulGateway> leaderRetriever,
+ final Time timeout,
+ final Map<String, String> responseHeaders) {
+ super(
+ leaderRetriever,
+ timeout,
+ responseHeaders,
+ CheckpointTriggerHeaders.getInstance());
+ }
+
+ protected AsynchronousJobOperationKey createOperationKey(
+ final HandlerRequest<CheckpointTriggerRequestBody> request) {
+ final JobID jobId = request.getPathParameter(JobIDPathParameter.class);
+ return AsynchronousJobOperationKey.of(
+ request.getRequestBody().getTriggerId().orElseGet(TriggerId::new), jobId);
+ }
+
+ @Override
+ protected CompletableFuture<TriggerResponse> handleRequest(
+ @Nonnull HandlerRequest<CheckpointTriggerRequestBody> request,
+ @Nonnull RestfulGateway gateway)
+ throws RestHandlerException {
+ final AsynchronousJobOperationKey operationKey = createOperationKey(request);
+
+ return gateway.triggerCheckpoint(
+ operationKey,
+ request.getRequestBody().getCheckpointType(),
+ RpcUtils.INF_TIMEOUT)
Review Comment:
Please file a ticket to remove this from this handler and the savepoint handler.
Or at least document why an infinite timeout is reasonable. (it usually isn't, and I suspect this just wasn't caught in the review)
##########
flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterGateway.java:
##########
@@ -218,13 +220,27 @@ CompletableFuture<String> triggerSavepoint(
final SavepointFormatType formatType,
@RpcTimeout final Time timeout);
+ /**
+ * Triggers taking a checkpoint of the executed job.
+ *
+ * @param checkpointType to determine how checkpoint should be taken or null if the existing
+ * checkpoint property should be used
Review Comment:
```suggestion
* @param checkpointType to determine how checkpoint should be taken
```
##########
flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/checkpoints/CheckpointTriggerHandlersTest.java:
##########
@@ -0,0 +1,349 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.rest.handler.job.checkpoints;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.core.execution.CheckpointType;
+import org.apache.flink.runtime.dispatcher.UnknownOperationKeyException;
+import org.apache.flink.runtime.messages.Acknowledge;
+import org.apache.flink.runtime.rest.RestMatchers;
+import org.apache.flink.runtime.rest.handler.HandlerRequest;
+import org.apache.flink.runtime.rest.handler.HandlerRequestException;
+import org.apache.flink.runtime.rest.handler.RestHandlerException;
+import org.apache.flink.runtime.rest.handler.async.AsynchronousOperationResult;
+import org.apache.flink.runtime.rest.handler.async.OperationResult;
+import org.apache.flink.runtime.rest.handler.job.AsynchronousJobOperationKey;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.JobIDPathParameter;
+import org.apache.flink.runtime.rest.messages.TriggerId;
+import org.apache.flink.runtime.rest.messages.TriggerIdPathParameter;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerInfo;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerMessageParameters;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerRequestBody;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerStatusMessageParameters;
+import org.apache.flink.runtime.rest.messages.queue.QueueStatus;
+import org.apache.flink.runtime.webmonitor.RestfulGateway;
+import org.apache.flink.runtime.webmonitor.TestingRestfulGateway;
+import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
+import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.concurrent.FutureUtils;
+
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/** Test for {@link CheckpointTriggerHandlers}. */
+public class CheckpointTriggerHandlersTest extends TestLogger {
+
+ private static final Time TIMEOUT = Time.seconds(10);
+
+ private static final JobID JOB_ID = new JobID();
+
+ private static final Long COMPLETED_CHECKPOINT_ID = 123456L;
+
+ private static CheckpointTriggerHandlers.CheckpointTriggerHandler checkpointTriggerHandler;
+
+ private static CheckpointTriggerHandlers.CheckpointTriggerStatusHandler
+ checkpointTriggerStatusHandler;
+
+ @BeforeAll
+ static void setUp() throws Exception {
+ GatewayRetriever<RestfulGateway> leaderRetriever =
+ () -> CompletableFuture.completedFuture(null);
+
+ checkpointTriggerHandler =
+ new CheckpointTriggerHandlers.CheckpointTriggerHandler(
+ leaderRetriever, TIMEOUT, Collections.emptyMap());
+
+ checkpointTriggerStatusHandler =
+ new CheckpointTriggerHandlers.CheckpointTriggerStatusHandler(
+ leaderRetriever, TIMEOUT, Collections.emptyMap());
+ }
+
+ @Test
+ public void testCheckpointTriggerCompletedSuccessfully() throws Exception {
+ final OperationResult<Long> successfulResult =
+ OperationResult.success(COMPLETED_CHECKPOINT_ID);
+ final CompletableFuture<CheckpointType> checkpointPropertiesFuture =
+ new CompletableFuture<>();
+
+ final AtomicReference<AsynchronousJobOperationKey> keyReference = new AtomicReference<>();
+ final TestingRestfulGateway testingRestfulGateway =
+ new TestingRestfulGateway.Builder()
+ .setTriggerCheckpointFunction(
+ (AsynchronousJobOperationKey key,
+ CheckpointType checkpointType) -> {
+ keyReference.set(key);
+ checkpointPropertiesFuture.complete(checkpointType);
+ return CompletableFuture.completedFuture(Acknowledge.get());
+ })
+ .setGetCheckpointStatusFunction(
+ (AsynchronousJobOperationKey operationKey) -> {
+ if (operationKey.equals(keyReference.get())) {
+ return CompletableFuture.completedFuture(successfulResult);
+ }
+ throw new RuntimeException(
+ "Expected operation key "
+ + keyReference.get()
+ + ", but received "
+ + operationKey);
+ })
+ .build();
+
+ final CheckpointType checkpointType = CheckpointType.FULL;
+
+ final TriggerId triggerId =
+ checkpointTriggerHandler
+ .handleRequest(
+ triggerCheckpointRequest(checkpointType, null),
+ testingRestfulGateway)
+ .get()
+ .getTriggerId();
+
+ final AsynchronousOperationResult<CheckpointTriggerInfo> checkpointTriggerResponseBody =
+ checkpointTriggerStatusHandler
+ .handleRequest(
+ checkpointTriggerStatusRequest(triggerId), testingRestfulGateway)
+ .get();
+
+ assertEquals(checkpointTriggerResponseBody.queueStatus().getId(), QueueStatus.Id.COMPLETED);
+ assertNotNull(checkpointTriggerResponseBody.resource());
+ assertEquals(
+ checkpointTriggerResponseBody.resource().getCheckpointId(),
+ COMPLETED_CHECKPOINT_ID);
+ assertEquals(checkpointPropertiesFuture.get(), CheckpointType.FULL);
+ }
+
+ @Test
+ public void testTriggerCheckpointNoCheckpointType() throws Exception {
+ final OperationResult<Long> successfulResult =
+ OperationResult.success(COMPLETED_CHECKPOINT_ID);
+ final CompletableFuture<CheckpointType> checkpointTypeFuture = new CompletableFuture<>();
+
+ final AtomicReference<AsynchronousJobOperationKey> keyReference = new AtomicReference<>();
+ final TestingRestfulGateway testingRestfulGateway =
+ new TestingRestfulGateway.Builder()
+ .setTriggerCheckpointFunction(
+ (AsynchronousJobOperationKey key,
+ CheckpointType checkpointType) -> {
+ keyReference.set(key);
+ checkpointTypeFuture.complete(checkpointType);
+ return CompletableFuture.completedFuture(Acknowledge.get());
+ })
+ .setGetCheckpointStatusFunction(
+ (AsynchronousJobOperationKey operationKey) -> {
+ if (operationKey.equals(keyReference.get())) {
+ return CompletableFuture.completedFuture(successfulResult);
+ }
+ throw new RuntimeException(
+ "Expected operation key "
+ + keyReference.get()
+ + ", but received "
+ + operationKey);
+ })
+ .build();
+
+ final TriggerId triggerId =
+ checkpointTriggerHandler
+ .handleRequest(triggerCheckpointRequest(null, null), testingRestfulGateway)
+ .get()
+ .getTriggerId();
+
+ AsynchronousOperationResult<CheckpointTriggerInfo> checkpointTriggerResponseBody;
+ checkpointTriggerResponseBody =
+ checkpointTriggerStatusHandler
+ .handleRequest(
+ checkpointTriggerStatusRequest(triggerId), testingRestfulGateway)
+ .get();
+
+ assertEquals(checkpointTriggerResponseBody.queueStatus().getId(), QueueStatus.Id.COMPLETED);
+ assertNotNull(checkpointTriggerResponseBody.resource());
+ assertEquals(
+ checkpointTriggerResponseBody.resource().getCheckpointId(),
+ COMPLETED_CHECKPOINT_ID);
+ assertEquals(checkpointTypeFuture.get(), CheckpointType.DEFAULT);
+ }
+
+ @Test
+ public void testCheckpointCompletedWithException() throws Exception {
+ final OperationResult<Long> failedResult =
+ OperationResult.failure(new RuntimeException("expected"));
+
+ final AtomicReference<AsynchronousJobOperationKey> keyReference = new AtomicReference<>();
+ final TestingRestfulGateway testingRestfulGateway =
+ new TestingRestfulGateway.Builder()
+ .setTriggerCheckpointFunction(
+ (AsynchronousJobOperationKey key,
+ CheckpointType checkpointType) -> {
+ keyReference.set(key);
+ return CompletableFuture.completedFuture(Acknowledge.get());
+ })
+ .setGetCheckpointStatusFunction(
+ (AsynchronousJobOperationKey operationKey) -> {
+ if (operationKey.equals(keyReference.get())) {
+ return CompletableFuture.completedFuture(failedResult);
+ }
+ throw new RuntimeException(
+ "Expected operation key "
+ + keyReference.get()
+ + ", but received "
+ + operationKey);
+ })
+ .build();
+
+ final TriggerId triggerId =
+ checkpointTriggerHandler
+ .handleRequest(triggerCheckpointRequest(null, null), testingRestfulGateway)
+ .get()
+ .getTriggerId();
+
+ AsynchronousOperationResult<CheckpointTriggerInfo> checkpointTriggerResponseBody;
+ checkpointTriggerResponseBody =
+ checkpointTriggerStatusHandler
+ .handleRequest(
+ checkpointTriggerStatusRequest(triggerId), testingRestfulGateway)
+ .get();
+
+ assertEquals(checkpointTriggerResponseBody.queueStatus().getId(), QueueStatus.Id.COMPLETED);
+ assertNotNull(checkpointTriggerResponseBody.resource());
+ assertNotNull(checkpointTriggerResponseBody.resource().getFailureCause());
+
+ final Throwable checkpointError =
+ checkpointTriggerResponseBody
+ .resource()
+ .getFailureCause()
+ .deserializeError(ClassLoader.getSystemClassLoader());
+ assertThat(checkpointError.getMessage()).matches("expected");
+ assertInstanceOf(RuntimeException.class, checkpointError);
+ }
+
+ @Test
+ public void testProvidedTriggerId() throws Exception {
+ final OperationResult<Long> successfulResult =
+ OperationResult.success(COMPLETED_CHECKPOINT_ID);
+ final AtomicReference<AsynchronousJobOperationKey> keyReference = new AtomicReference<>();
+ final TestingRestfulGateway testingRestfulGateway =
+ new TestingRestfulGateway.Builder()
+ .setTriggerCheckpointFunction(
+ (AsynchronousJobOperationKey key,
+ CheckpointType checkpointType) -> {
+ keyReference.set(key);
+ return CompletableFuture.completedFuture(Acknowledge.get());
+ })
+ .setGetCheckpointStatusFunction(
+ (AsynchronousJobOperationKey operationKey) -> {
+ if (operationKey.equals(keyReference.get())) {
+ return CompletableFuture.completedFuture(successfulResult);
+ }
+ throw new RuntimeException(
+ "Expected operation key "
+ + keyReference.get()
+ + ", but received "
+ + operationKey);
+ })
+ .build();
+
+ final TriggerId providedTriggerId = new TriggerId();
+
+ final TriggerId returnedTriggerId =
+ checkpointTriggerHandler
+ .handleRequest(
+ triggerCheckpointRequest(CheckpointType.FULL, providedTriggerId),
+ testingRestfulGateway)
+ .get()
+ .getTriggerId();
+
+ assertEquals(providedTriggerId, returnedTriggerId);
+
+ AsynchronousOperationResult<CheckpointTriggerInfo> checkpointTriggerResponseBody;
+ checkpointTriggerResponseBody =
+ checkpointTriggerStatusHandler
+ .handleRequest(
+ checkpointTriggerStatusRequest(providedTriggerId),
+ testingRestfulGateway)
+ .get();
+
+ assertEquals(checkpointTriggerResponseBody.queueStatus().getId(), QueueStatus.Id.COMPLETED);
+ assertNotNull(checkpointTriggerResponseBody.resource());
Review Comment:
Please use assertj for all assertions.
That usually looks like this `assertThat(<objectToAssert>).<some assertion>`
##########
flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/checkpoints/CheckpointTriggerHandlersTest.java:
##########
@@ -0,0 +1,349 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.rest.handler.job.checkpoints;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.core.execution.CheckpointType;
+import org.apache.flink.runtime.dispatcher.UnknownOperationKeyException;
+import org.apache.flink.runtime.messages.Acknowledge;
+import org.apache.flink.runtime.rest.RestMatchers;
+import org.apache.flink.runtime.rest.handler.HandlerRequest;
+import org.apache.flink.runtime.rest.handler.HandlerRequestException;
+import org.apache.flink.runtime.rest.handler.RestHandlerException;
+import org.apache.flink.runtime.rest.handler.async.AsynchronousOperationResult;
+import org.apache.flink.runtime.rest.handler.async.OperationResult;
+import org.apache.flink.runtime.rest.handler.job.AsynchronousJobOperationKey;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.JobIDPathParameter;
+import org.apache.flink.runtime.rest.messages.TriggerId;
+import org.apache.flink.runtime.rest.messages.TriggerIdPathParameter;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerInfo;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerMessageParameters;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerRequestBody;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerStatusMessageParameters;
+import org.apache.flink.runtime.rest.messages.queue.QueueStatus;
+import org.apache.flink.runtime.webmonitor.RestfulGateway;
+import org.apache.flink.runtime.webmonitor.TestingRestfulGateway;
+import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
+import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.concurrent.FutureUtils;
+
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/** Test for {@link CheckpointTriggerHandlers}. */
+public class CheckpointTriggerHandlersTest extends TestLogger {
+
+ private static final Time TIMEOUT = Time.seconds(10);
+
+ private static final JobID JOB_ID = new JobID();
+
+ private static final Long COMPLETED_CHECKPOINT_ID = 123456L;
+
+ private static CheckpointTriggerHandlers.CheckpointTriggerHandler checkpointTriggerHandler;
+
+ private static CheckpointTriggerHandlers.CheckpointTriggerStatusHandler
+ checkpointTriggerStatusHandler;
+
+ @BeforeAll
+ static void setUp() throws Exception {
+ GatewayRetriever<RestfulGateway> leaderRetriever =
+ () -> CompletableFuture.completedFuture(null);
+
+ checkpointTriggerHandler =
+ new CheckpointTriggerHandlers.CheckpointTriggerHandler(
+ leaderRetriever, TIMEOUT, Collections.emptyMap());
+
+ checkpointTriggerStatusHandler =
+ new CheckpointTriggerHandlers.CheckpointTriggerStatusHandler(
+ leaderRetriever, TIMEOUT, Collections.emptyMap());
+ }
+
+ @Test
+ public void testCheckpointTriggerCompletedSuccessfully() throws Exception {
+ final OperationResult<Long> successfulResult =
+ OperationResult.success(COMPLETED_CHECKPOINT_ID);
+ final CompletableFuture<CheckpointType> checkpointPropertiesFuture =
+ new CompletableFuture<>();
+
+ final AtomicReference<AsynchronousJobOperationKey> keyReference = new AtomicReference<>();
+ final TestingRestfulGateway testingRestfulGateway =
+ new TestingRestfulGateway.Builder()
+ .setTriggerCheckpointFunction(
+ (AsynchronousJobOperationKey key,
+ CheckpointType checkpointType) -> {
+ keyReference.set(key);
+ checkpointPropertiesFuture.complete(checkpointType);
+ return CompletableFuture.completedFuture(Acknowledge.get());
+ })
+ .setGetCheckpointStatusFunction(
+ (AsynchronousJobOperationKey operationKey) -> {
+ if (operationKey.equals(keyReference.get())) {
+ return CompletableFuture.completedFuture(successfulResult);
+ }
+ throw new RuntimeException(
+ "Expected operation key "
+ + keyReference.get()
+ + ", but received "
+ + operationKey);
+ })
+ .build();
+
+ final CheckpointType checkpointType = CheckpointType.FULL;
+
+ final TriggerId triggerId =
+ checkpointTriggerHandler
+ .handleRequest(
+ triggerCheckpointRequest(checkpointType, null),
+ testingRestfulGateway)
+ .get()
+ .getTriggerId();
+
+ final AsynchronousOperationResult<CheckpointTriggerInfo> checkpointTriggerResponseBody =
+ checkpointTriggerStatusHandler
+ .handleRequest(
+ checkpointTriggerStatusRequest(triggerId), testingRestfulGateway)
+ .get();
+
+ assertEquals(checkpointTriggerResponseBody.queueStatus().getId(), QueueStatus.Id.COMPLETED);
+ assertNotNull(checkpointTriggerResponseBody.resource());
+ assertEquals(
+ checkpointTriggerResponseBody.resource().getCheckpointId(),
+ COMPLETED_CHECKPOINT_ID);
+ assertEquals(checkpointPropertiesFuture.get(), CheckpointType.FULL);
+ }
+
+ @Test
+ public void testTriggerCheckpointNoCheckpointType() throws Exception {
+ final OperationResult<Long> successfulResult =
+ OperationResult.success(COMPLETED_CHECKPOINT_ID);
+ final CompletableFuture<CheckpointType> checkpointTypeFuture = new CompletableFuture<>();
+
+ final AtomicReference<AsynchronousJobOperationKey> keyReference = new AtomicReference<>();
+ final TestingRestfulGateway testingRestfulGateway =
+ new TestingRestfulGateway.Builder()
+ .setTriggerCheckpointFunction(
+ (AsynchronousJobOperationKey key,
+ CheckpointType checkpointType) -> {
+ keyReference.set(key);
+ checkpointTypeFuture.complete(checkpointType);
+ return CompletableFuture.completedFuture(Acknowledge.get());
+ })
+ .setGetCheckpointStatusFunction(
+ (AsynchronousJobOperationKey operationKey) -> {
+ if (operationKey.equals(keyReference.get())) {
+ return CompletableFuture.completedFuture(successfulResult);
+ }
+ throw new RuntimeException(
+ "Expected operation key "
+ + keyReference.get()
+ + ", but received "
+ + operationKey);
+ })
+ .build();
+
+ final TriggerId triggerId =
+ checkpointTriggerHandler
+ .handleRequest(triggerCheckpointRequest(null, null), testingRestfulGateway)
+ .get()
+ .getTriggerId();
+
+ AsynchronousOperationResult<CheckpointTriggerInfo> checkpointTriggerResponseBody;
+ checkpointTriggerResponseBody =
+ checkpointTriggerStatusHandler
+ .handleRequest(
+ checkpointTriggerStatusRequest(triggerId), testingRestfulGateway)
+ .get();
+
+ assertEquals(checkpointTriggerResponseBody.queueStatus().getId(), QueueStatus.Id.COMPLETED);
+ assertNotNull(checkpointTriggerResponseBody.resource());
+ assertEquals(
+ checkpointTriggerResponseBody.resource().getCheckpointId(),
+ COMPLETED_CHECKPOINT_ID);
+ assertEquals(checkpointTypeFuture.get(), CheckpointType.DEFAULT);
+ }
+
+ @Test
+ public void testCheckpointCompletedWithException() throws Exception {
+ final OperationResult<Long> failedResult =
+ OperationResult.failure(new RuntimeException("expected"));
+
+ final AtomicReference<AsynchronousJobOperationKey> keyReference = new AtomicReference<>();
+ final TestingRestfulGateway testingRestfulGateway =
+ new TestingRestfulGateway.Builder()
+ .setTriggerCheckpointFunction(
+ (AsynchronousJobOperationKey key,
+ CheckpointType checkpointType) -> {
+ keyReference.set(key);
+ return CompletableFuture.completedFuture(Acknowledge.get());
+ })
+ .setGetCheckpointStatusFunction(
+ (AsynchronousJobOperationKey operationKey) -> {
+ if (operationKey.equals(keyReference.get())) {
+ return CompletableFuture.completedFuture(failedResult);
+ }
+ throw new RuntimeException(
+ "Expected operation key "
+ + keyReference.get()
+ + ", but received "
+ + operationKey);
+ })
+ .build();
+
+ final TriggerId triggerId =
+ checkpointTriggerHandler
+ .handleRequest(triggerCheckpointRequest(null, null), testingRestfulGateway)
+ .get()
+ .getTriggerId();
+
+ AsynchronousOperationResult<CheckpointTriggerInfo> checkpointTriggerResponseBody;
+ checkpointTriggerResponseBody =
+ checkpointTriggerStatusHandler
+ .handleRequest(
+ checkpointTriggerStatusRequest(triggerId), testingRestfulGateway)
+ .get();
+
+ assertEquals(checkpointTriggerResponseBody.queueStatus().getId(), QueueStatus.Id.COMPLETED);
+ assertNotNull(checkpointTriggerResponseBody.resource());
+ assertNotNull(checkpointTriggerResponseBody.resource().getFailureCause());
+
+ final Throwable checkpointError =
+ checkpointTriggerResponseBody
+ .resource()
+ .getFailureCause()
+ .deserializeError(ClassLoader.getSystemClassLoader());
+ assertThat(checkpointError.getMessage()).matches("expected");
+ assertInstanceOf(RuntimeException.class, checkpointError);
+ }
+
+ @Test
+ public void testProvidedTriggerId() throws Exception {
Review Comment:
```suggestion
void testProvidedTriggerId() throws Exception {
```
##########
flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/checkpoints/CheckpointTriggerStatusHeaders.java:
##########
@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.rest.messages.checkpoints;
+
+import org.apache.flink.runtime.rest.HttpMethodWrapper;
+import org.apache.flink.runtime.rest.handler.async.AsynchronousOperationStatusMessageHeaders;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.JobIDPathParameter;
+import org.apache.flink.runtime.rest.messages.TriggerIdPathParameter;
+
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+/** These headers define the protocol for triggering a checkpoint. */
+public class CheckpointTriggerStatusHeaders
+ extends AsynchronousOperationStatusMessageHeaders<
+ CheckpointTriggerInfo, CheckpointTriggerStatusMessageParameters> {
Review Comment:
I would remove "Trigger" from class names that are related to the status querying.
This makes this whole thing ambiguous; because based in the name this returns the status of the _trigger_, and not of the _checkpoint_.
##########
flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/checkpoints/CheckpointTriggerHandlersTest.java:
##########
@@ -0,0 +1,349 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.rest.handler.job.checkpoints;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.core.execution.CheckpointType;
+import org.apache.flink.runtime.dispatcher.UnknownOperationKeyException;
+import org.apache.flink.runtime.messages.Acknowledge;
+import org.apache.flink.runtime.rest.RestMatchers;
+import org.apache.flink.runtime.rest.handler.HandlerRequest;
+import org.apache.flink.runtime.rest.handler.HandlerRequestException;
+import org.apache.flink.runtime.rest.handler.RestHandlerException;
+import org.apache.flink.runtime.rest.handler.async.AsynchronousOperationResult;
+import org.apache.flink.runtime.rest.handler.async.OperationResult;
+import org.apache.flink.runtime.rest.handler.job.AsynchronousJobOperationKey;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.JobIDPathParameter;
+import org.apache.flink.runtime.rest.messages.TriggerId;
+import org.apache.flink.runtime.rest.messages.TriggerIdPathParameter;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerInfo;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerMessageParameters;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerRequestBody;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerStatusMessageParameters;
+import org.apache.flink.runtime.rest.messages.queue.QueueStatus;
+import org.apache.flink.runtime.webmonitor.RestfulGateway;
+import org.apache.flink.runtime.webmonitor.TestingRestfulGateway;
+import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
+import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.concurrent.FutureUtils;
+
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/** Test for {@link CheckpointTriggerHandlers}. */
+public class CheckpointTriggerHandlersTest extends TestLogger {
+
+ private static final Time TIMEOUT = Time.seconds(10);
+
+ private static final JobID JOB_ID = new JobID();
+
+ private static final Long COMPLETED_CHECKPOINT_ID = 123456L;
+
+ private static CheckpointTriggerHandlers.CheckpointTriggerHandler checkpointTriggerHandler;
+
+ private static CheckpointTriggerHandlers.CheckpointTriggerStatusHandler
+ checkpointTriggerStatusHandler;
+
+ @BeforeAll
+ static void setUp() throws Exception {
+ GatewayRetriever<RestfulGateway> leaderRetriever =
+ () -> CompletableFuture.completedFuture(null);
+
+ checkpointTriggerHandler =
+ new CheckpointTriggerHandlers.CheckpointTriggerHandler(
+ leaderRetriever, TIMEOUT, Collections.emptyMap());
+
+ checkpointTriggerStatusHandler =
+ new CheckpointTriggerHandlers.CheckpointTriggerStatusHandler(
+ leaderRetriever, TIMEOUT, Collections.emptyMap());
+ }
+
+ @Test
+ public void testCheckpointTriggerCompletedSuccessfully() throws Exception {
+ final OperationResult<Long> successfulResult =
+ OperationResult.success(COMPLETED_CHECKPOINT_ID);
+ final CompletableFuture<CheckpointType> checkpointPropertiesFuture =
+ new CompletableFuture<>();
+
+ final AtomicReference<AsynchronousJobOperationKey> keyReference = new AtomicReference<>();
+ final TestingRestfulGateway testingRestfulGateway =
+ new TestingRestfulGateway.Builder()
+ .setTriggerCheckpointFunction(
+ (AsynchronousJobOperationKey key,
+ CheckpointType checkpointType) -> {
+ keyReference.set(key);
+ checkpointPropertiesFuture.complete(checkpointType);
+ return CompletableFuture.completedFuture(Acknowledge.get());
+ })
+ .setGetCheckpointStatusFunction(
+ (AsynchronousJobOperationKey operationKey) -> {
+ if (operationKey.equals(keyReference.get())) {
+ return CompletableFuture.completedFuture(successfulResult);
+ }
+ throw new RuntimeException(
+ "Expected operation key "
+ + keyReference.get()
+ + ", but received "
+ + operationKey);
+ })
+ .build();
+
+ final CheckpointType checkpointType = CheckpointType.FULL;
+
+ final TriggerId triggerId =
+ checkpointTriggerHandler
+ .handleRequest(
+ triggerCheckpointRequest(checkpointType, null),
+ testingRestfulGateway)
+ .get()
+ .getTriggerId();
+
+ final AsynchronousOperationResult<CheckpointTriggerInfo> checkpointTriggerResponseBody =
+ checkpointTriggerStatusHandler
+ .handleRequest(
+ checkpointTriggerStatusRequest(triggerId), testingRestfulGateway)
+ .get();
+
+ assertEquals(checkpointTriggerResponseBody.queueStatus().getId(), QueueStatus.Id.COMPLETED);
+ assertNotNull(checkpointTriggerResponseBody.resource());
+ assertEquals(
+ checkpointTriggerResponseBody.resource().getCheckpointId(),
+ COMPLETED_CHECKPOINT_ID);
+ assertEquals(checkpointPropertiesFuture.get(), CheckpointType.FULL);
+ }
+
+ @Test
+ public void testTriggerCheckpointNoCheckpointType() throws Exception {
+ final OperationResult<Long> successfulResult =
+ OperationResult.success(COMPLETED_CHECKPOINT_ID);
+ final CompletableFuture<CheckpointType> checkpointTypeFuture = new CompletableFuture<>();
+
+ final AtomicReference<AsynchronousJobOperationKey> keyReference = new AtomicReference<>();
+ final TestingRestfulGateway testingRestfulGateway =
+ new TestingRestfulGateway.Builder()
+ .setTriggerCheckpointFunction(
+ (AsynchronousJobOperationKey key,
+ CheckpointType checkpointType) -> {
+ keyReference.set(key);
+ checkpointTypeFuture.complete(checkpointType);
+ return CompletableFuture.completedFuture(Acknowledge.get());
+ })
+ .setGetCheckpointStatusFunction(
+ (AsynchronousJobOperationKey operationKey) -> {
+ if (operationKey.equals(keyReference.get())) {
+ return CompletableFuture.completedFuture(successfulResult);
+ }
+ throw new RuntimeException(
+ "Expected operation key "
+ + keyReference.get()
+ + ", but received "
+ + operationKey);
+ })
+ .build();
+
+ final TriggerId triggerId =
+ checkpointTriggerHandler
+ .handleRequest(triggerCheckpointRequest(null, null), testingRestfulGateway)
+ .get()
+ .getTriggerId();
+
+ AsynchronousOperationResult<CheckpointTriggerInfo> checkpointTriggerResponseBody;
+ checkpointTriggerResponseBody =
+ checkpointTriggerStatusHandler
+ .handleRequest(
+ checkpointTriggerStatusRequest(triggerId), testingRestfulGateway)
+ .get();
+
+ assertEquals(checkpointTriggerResponseBody.queueStatus().getId(), QueueStatus.Id.COMPLETED);
+ assertNotNull(checkpointTriggerResponseBody.resource());
+ assertEquals(
+ checkpointTriggerResponseBody.resource().getCheckpointId(),
+ COMPLETED_CHECKPOINT_ID);
+ assertEquals(checkpointTypeFuture.get(), CheckpointType.DEFAULT);
+ }
+
+ @Test
+ public void testCheckpointCompletedWithException() throws Exception {
+ final OperationResult<Long> failedResult =
+ OperationResult.failure(new RuntimeException("expected"));
+
+ final AtomicReference<AsynchronousJobOperationKey> keyReference = new AtomicReference<>();
+ final TestingRestfulGateway testingRestfulGateway =
+ new TestingRestfulGateway.Builder()
+ .setTriggerCheckpointFunction(
+ (AsynchronousJobOperationKey key,
+ CheckpointType checkpointType) -> {
+ keyReference.set(key);
+ return CompletableFuture.completedFuture(Acknowledge.get());
+ })
+ .setGetCheckpointStatusFunction(
+ (AsynchronousJobOperationKey operationKey) -> {
+ if (operationKey.equals(keyReference.get())) {
+ return CompletableFuture.completedFuture(failedResult);
+ }
+ throw new RuntimeException(
+ "Expected operation key "
+ + keyReference.get()
+ + ", but received "
+ + operationKey);
+ })
+ .build();
+
+ final TriggerId triggerId =
+ checkpointTriggerHandler
+ .handleRequest(triggerCheckpointRequest(null, null), testingRestfulGateway)
+ .get()
+ .getTriggerId();
+
+ AsynchronousOperationResult<CheckpointTriggerInfo> checkpointTriggerResponseBody;
+ checkpointTriggerResponseBody =
+ checkpointTriggerStatusHandler
+ .handleRequest(
+ checkpointTriggerStatusRequest(triggerId), testingRestfulGateway)
+ .get();
+
+ assertEquals(checkpointTriggerResponseBody.queueStatus().getId(), QueueStatus.Id.COMPLETED);
+ assertNotNull(checkpointTriggerResponseBody.resource());
+ assertNotNull(checkpointTriggerResponseBody.resource().getFailureCause());
+
+ final Throwable checkpointError =
+ checkpointTriggerResponseBody
+ .resource()
+ .getFailureCause()
+ .deserializeError(ClassLoader.getSystemClassLoader());
+ assertThat(checkpointError.getMessage()).matches("expected");
+ assertInstanceOf(RuntimeException.class, checkpointError);
+ }
+
+ @Test
+ public void testProvidedTriggerId() throws Exception {
+ final OperationResult<Long> successfulResult =
+ OperationResult.success(COMPLETED_CHECKPOINT_ID);
+ final AtomicReference<AsynchronousJobOperationKey> keyReference = new AtomicReference<>();
+ final TestingRestfulGateway testingRestfulGateway =
+ new TestingRestfulGateway.Builder()
+ .setTriggerCheckpointFunction(
+ (AsynchronousJobOperationKey key,
+ CheckpointType checkpointType) -> {
+ keyReference.set(key);
+ return CompletableFuture.completedFuture(Acknowledge.get());
+ })
+ .setGetCheckpointStatusFunction(
+ (AsynchronousJobOperationKey operationKey) -> {
+ if (operationKey.equals(keyReference.get())) {
+ return CompletableFuture.completedFuture(successfulResult);
+ }
+ throw new RuntimeException(
+ "Expected operation key "
+ + keyReference.get()
+ + ", but received "
+ + operationKey);
+ })
+ .build();
+
+ final TriggerId providedTriggerId = new TriggerId();
+
+ final TriggerId returnedTriggerId =
+ checkpointTriggerHandler
+ .handleRequest(
+ triggerCheckpointRequest(CheckpointType.FULL, providedTriggerId),
+ testingRestfulGateway)
+ .get()
+ .getTriggerId();
+
+ assertEquals(providedTriggerId, returnedTriggerId);
+
+ AsynchronousOperationResult<CheckpointTriggerInfo> checkpointTriggerResponseBody;
+ checkpointTriggerResponseBody =
+ checkpointTriggerStatusHandler
+ .handleRequest(
+ checkpointTriggerStatusRequest(providedTriggerId),
+ testingRestfulGateway)
+ .get();
+
+ assertEquals(checkpointTriggerResponseBody.queueStatus().getId(), QueueStatus.Id.COMPLETED);
+ assertNotNull(checkpointTriggerResponseBody.resource());
+ assertEquals(
+ checkpointTriggerResponseBody.resource().getCheckpointId(),
+ COMPLETED_CHECKPOINT_ID);
+ }
+
+ @Test
+ public void testQueryStatusOfUnknownOperationReturnsError()
+ throws HandlerRequestException, RestHandlerException, ExecutionException,
+ InterruptedException {
+
+ final TestingRestfulGateway testingRestfulGateway =
+ new TestingRestfulGateway.Builder()
+ .setGetCheckpointStatusFunction(
+ key ->
+ FutureUtils.completedExceptionally(
+ new UnknownOperationKeyException(key)))
+ .build();
+
+ final CompletableFuture<AsynchronousOperationResult<CheckpointTriggerInfo>> statusFuture =
+ checkpointTriggerStatusHandler.handleRequest(
+ checkpointTriggerStatusRequest(new TriggerId()), testingRestfulGateway);
+
+ assertTrue(
+ RestMatchers.respondsWithError(HttpResponseStatus.NOT_FOUND).matches(statusFuture));
+ }
+
+ private static HandlerRequest<CheckpointTriggerRequestBody> triggerCheckpointRequest(
+ final CheckpointType checkpointType, final TriggerId triggerId)
Review Comment:
```suggestion
final CheckpointType checkpointType, @Nullable final TriggerId triggerId)
```
##########
flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/checkpoints/CheckpointTriggerHandlersTest.java:
##########
@@ -0,0 +1,349 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.rest.handler.job.checkpoints;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.core.execution.CheckpointType;
+import org.apache.flink.runtime.dispatcher.UnknownOperationKeyException;
+import org.apache.flink.runtime.messages.Acknowledge;
+import org.apache.flink.runtime.rest.RestMatchers;
+import org.apache.flink.runtime.rest.handler.HandlerRequest;
+import org.apache.flink.runtime.rest.handler.HandlerRequestException;
+import org.apache.flink.runtime.rest.handler.RestHandlerException;
+import org.apache.flink.runtime.rest.handler.async.AsynchronousOperationResult;
+import org.apache.flink.runtime.rest.handler.async.OperationResult;
+import org.apache.flink.runtime.rest.handler.job.AsynchronousJobOperationKey;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.JobIDPathParameter;
+import org.apache.flink.runtime.rest.messages.TriggerId;
+import org.apache.flink.runtime.rest.messages.TriggerIdPathParameter;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerInfo;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerMessageParameters;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerRequestBody;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerStatusMessageParameters;
+import org.apache.flink.runtime.rest.messages.queue.QueueStatus;
+import org.apache.flink.runtime.webmonitor.RestfulGateway;
+import org.apache.flink.runtime.webmonitor.TestingRestfulGateway;
+import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
+import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.concurrent.FutureUtils;
+
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/** Test for {@link CheckpointTriggerHandlers}. */
+public class CheckpointTriggerHandlersTest extends TestLogger {
+
+ private static final Time TIMEOUT = Time.seconds(10);
+
+ private static final JobID JOB_ID = new JobID();
+
+ private static final Long COMPLETED_CHECKPOINT_ID = 123456L;
+
+ private static CheckpointTriggerHandlers.CheckpointTriggerHandler checkpointTriggerHandler;
+
+ private static CheckpointTriggerHandlers.CheckpointTriggerStatusHandler
+ checkpointTriggerStatusHandler;
+
+ @BeforeAll
+ static void setUp() throws Exception {
+ GatewayRetriever<RestfulGateway> leaderRetriever =
+ () -> CompletableFuture.completedFuture(null);
+
+ checkpointTriggerHandler =
+ new CheckpointTriggerHandlers.CheckpointTriggerHandler(
+ leaderRetriever, TIMEOUT, Collections.emptyMap());
+
+ checkpointTriggerStatusHandler =
+ new CheckpointTriggerHandlers.CheckpointTriggerStatusHandler(
+ leaderRetriever, TIMEOUT, Collections.emptyMap());
+ }
+
+ @Test
+ public void testCheckpointTriggerCompletedSuccessfully() throws Exception {
+ final OperationResult<Long> successfulResult =
+ OperationResult.success(COMPLETED_CHECKPOINT_ID);
+ final CompletableFuture<CheckpointType> checkpointPropertiesFuture =
+ new CompletableFuture<>();
+
+ final AtomicReference<AsynchronousJobOperationKey> keyReference = new AtomicReference<>();
+ final TestingRestfulGateway testingRestfulGateway =
+ new TestingRestfulGateway.Builder()
+ .setTriggerCheckpointFunction(
+ (AsynchronousJobOperationKey key,
+ CheckpointType checkpointType) -> {
+ keyReference.set(key);
+ checkpointPropertiesFuture.complete(checkpointType);
+ return CompletableFuture.completedFuture(Acknowledge.get());
+ })
+ .setGetCheckpointStatusFunction(
+ (AsynchronousJobOperationKey operationKey) -> {
+ if (operationKey.equals(keyReference.get())) {
+ return CompletableFuture.completedFuture(successfulResult);
+ }
+ throw new RuntimeException(
+ "Expected operation key "
+ + keyReference.get()
+ + ", but received "
+ + operationKey);
+ })
+ .build();
+
+ final CheckpointType checkpointType = CheckpointType.FULL;
+
+ final TriggerId triggerId =
+ checkpointTriggerHandler
+ .handleRequest(
+ triggerCheckpointRequest(checkpointType, null),
+ testingRestfulGateway)
+ .get()
+ .getTriggerId();
+
+ final AsynchronousOperationResult<CheckpointTriggerInfo> checkpointTriggerResponseBody =
+ checkpointTriggerStatusHandler
+ .handleRequest(
+ checkpointTriggerStatusRequest(triggerId), testingRestfulGateway)
+ .get();
+
+ assertEquals(checkpointTriggerResponseBody.queueStatus().getId(), QueueStatus.Id.COMPLETED);
+ assertNotNull(checkpointTriggerResponseBody.resource());
+ assertEquals(
+ checkpointTriggerResponseBody.resource().getCheckpointId(),
+ COMPLETED_CHECKPOINT_ID);
+ assertEquals(checkpointPropertiesFuture.get(), CheckpointType.FULL);
+ }
+
+ @Test
+ public void testTriggerCheckpointNoCheckpointType() throws Exception {
+ final OperationResult<Long> successfulResult =
+ OperationResult.success(COMPLETED_CHECKPOINT_ID);
+ final CompletableFuture<CheckpointType> checkpointTypeFuture = new CompletableFuture<>();
+
+ final AtomicReference<AsynchronousJobOperationKey> keyReference = new AtomicReference<>();
+ final TestingRestfulGateway testingRestfulGateway =
+ new TestingRestfulGateway.Builder()
+ .setTriggerCheckpointFunction(
+ (AsynchronousJobOperationKey key,
+ CheckpointType checkpointType) -> {
+ keyReference.set(key);
+ checkpointTypeFuture.complete(checkpointType);
+ return CompletableFuture.completedFuture(Acknowledge.get());
+ })
+ .setGetCheckpointStatusFunction(
+ (AsynchronousJobOperationKey operationKey) -> {
+ if (operationKey.equals(keyReference.get())) {
+ return CompletableFuture.completedFuture(successfulResult);
+ }
+ throw new RuntimeException(
+ "Expected operation key "
+ + keyReference.get()
+ + ", but received "
+ + operationKey);
+ })
+ .build();
+
+ final TriggerId triggerId =
+ checkpointTriggerHandler
+ .handleRequest(triggerCheckpointRequest(null, null), testingRestfulGateway)
+ .get()
+ .getTriggerId();
+
+ AsynchronousOperationResult<CheckpointTriggerInfo> checkpointTriggerResponseBody;
+ checkpointTriggerResponseBody =
+ checkpointTriggerStatusHandler
+ .handleRequest(
+ checkpointTriggerStatusRequest(triggerId), testingRestfulGateway)
+ .get();
+
+ assertEquals(checkpointTriggerResponseBody.queueStatus().getId(), QueueStatus.Id.COMPLETED);
+ assertNotNull(checkpointTriggerResponseBody.resource());
+ assertEquals(
+ checkpointTriggerResponseBody.resource().getCheckpointId(),
+ COMPLETED_CHECKPOINT_ID);
+ assertEquals(checkpointTypeFuture.get(), CheckpointType.DEFAULT);
+ }
+
+ @Test
+ public void testCheckpointCompletedWithException() throws Exception {
+ final OperationResult<Long> failedResult =
+ OperationResult.failure(new RuntimeException("expected"));
+
+ final AtomicReference<AsynchronousJobOperationKey> keyReference = new AtomicReference<>();
+ final TestingRestfulGateway testingRestfulGateway =
+ new TestingRestfulGateway.Builder()
+ .setTriggerCheckpointFunction(
+ (AsynchronousJobOperationKey key,
+ CheckpointType checkpointType) -> {
+ keyReference.set(key);
+ return CompletableFuture.completedFuture(Acknowledge.get());
+ })
+ .setGetCheckpointStatusFunction(
+ (AsynchronousJobOperationKey operationKey) -> {
+ if (operationKey.equals(keyReference.get())) {
+ return CompletableFuture.completedFuture(failedResult);
+ }
+ throw new RuntimeException(
+ "Expected operation key "
+ + keyReference.get()
+ + ", but received "
+ + operationKey);
+ })
+ .build();
+
+ final TriggerId triggerId =
+ checkpointTriggerHandler
+ .handleRequest(triggerCheckpointRequest(null, null), testingRestfulGateway)
+ .get()
+ .getTriggerId();
+
+ AsynchronousOperationResult<CheckpointTriggerInfo> checkpointTriggerResponseBody;
+ checkpointTriggerResponseBody =
+ checkpointTriggerStatusHandler
+ .handleRequest(
+ checkpointTriggerStatusRequest(triggerId), testingRestfulGateway)
+ .get();
+
+ assertEquals(checkpointTriggerResponseBody.queueStatus().getId(), QueueStatus.Id.COMPLETED);
+ assertNotNull(checkpointTriggerResponseBody.resource());
+ assertNotNull(checkpointTriggerResponseBody.resource().getFailureCause());
+
+ final Throwable checkpointError =
+ checkpointTriggerResponseBody
+ .resource()
+ .getFailureCause()
+ .deserializeError(ClassLoader.getSystemClassLoader());
+ assertThat(checkpointError.getMessage()).matches("expected");
+ assertInstanceOf(RuntimeException.class, checkpointError);
+ }
+
+ @Test
+ public void testProvidedTriggerId() throws Exception {
+ final OperationResult<Long> successfulResult =
+ OperationResult.success(COMPLETED_CHECKPOINT_ID);
+ final AtomicReference<AsynchronousJobOperationKey> keyReference = new AtomicReference<>();
+ final TestingRestfulGateway testingRestfulGateway =
+ new TestingRestfulGateway.Builder()
+ .setTriggerCheckpointFunction(
+ (AsynchronousJobOperationKey key,
+ CheckpointType checkpointType) -> {
+ keyReference.set(key);
+ return CompletableFuture.completedFuture(Acknowledge.get());
+ })
+ .setGetCheckpointStatusFunction(
+ (AsynchronousJobOperationKey operationKey) -> {
+ if (operationKey.equals(keyReference.get())) {
+ return CompletableFuture.completedFuture(successfulResult);
+ }
+ throw new RuntimeException(
+ "Expected operation key "
+ + keyReference.get()
+ + ", but received "
+ + operationKey);
+ })
+ .build();
+
+ final TriggerId providedTriggerId = new TriggerId();
+
+ final TriggerId returnedTriggerId =
+ checkpointTriggerHandler
+ .handleRequest(
+ triggerCheckpointRequest(CheckpointType.FULL, providedTriggerId),
+ testingRestfulGateway)
+ .get()
+ .getTriggerId();
+
+ assertEquals(providedTriggerId, returnedTriggerId);
+
+ AsynchronousOperationResult<CheckpointTriggerInfo> checkpointTriggerResponseBody;
+ checkpointTriggerResponseBody =
+ checkpointTriggerStatusHandler
+ .handleRequest(
+ checkpointTriggerStatusRequest(providedTriggerId),
+ testingRestfulGateway)
+ .get();
+
+ assertEquals(checkpointTriggerResponseBody.queueStatus().getId(), QueueStatus.Id.COMPLETED);
+ assertNotNull(checkpointTriggerResponseBody.resource());
+ assertEquals(
+ checkpointTriggerResponseBody.resource().getCheckpointId(),
+ COMPLETED_CHECKPOINT_ID);
+ }
+
+ @Test
+ public void testQueryStatusOfUnknownOperationReturnsError()
Review Comment:
```suggestion
void testQueryStatusOfUnknownOperationReturnsError()
```
##########
flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/checkpoints/CheckpointTriggerHandlers.java:
##########
@@ -0,0 +1,261 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.rest.handler.job.checkpoints;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.core.execution.CheckpointType;
+import org.apache.flink.runtime.dispatcher.UnknownOperationKeyException;
+import org.apache.flink.runtime.rest.handler.AbstractRestHandler;
+import org.apache.flink.runtime.rest.handler.HandlerRequest;
+import org.apache.flink.runtime.rest.handler.RestHandlerException;
+import org.apache.flink.runtime.rest.handler.async.AsynchronousOperationResult;
+import org.apache.flink.runtime.rest.handler.async.TriggerResponse;
+import org.apache.flink.runtime.rest.handler.job.AsynchronousJobOperationKey;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.JobIDPathParameter;
+import org.apache.flink.runtime.rest.messages.TriggerId;
+import org.apache.flink.runtime.rest.messages.TriggerIdPathParameter;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerHeaders;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerInfo;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerMessageParameters;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerRequestBody;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerStatusHeaders;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerStatusMessageParameters;
+import org.apache.flink.runtime.rpc.RpcUtils;
+import org.apache.flink.runtime.webmonitor.RestfulGateway;
+import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.SerializedThrowable;
+
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+import javax.annotation.Nonnull;
+
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+
+/**
+ * HTTP handlers for asynchronous triggering of checkpoints.
+ *
+ * <p>Drawing checkpoints is a potentially long-running operation. To avoid blocking HTTP
+ * connections, checkpoints must be drawn in two steps. First, an HTTP request is issued to trigger
+ * the checkpoint asynchronously. The request will be assigned a {@link TriggerId}, which is
+ * returned in the response body. Next, the returned {@link TriggerId} should be used to poll the
+ * status of the checkpoint until it is finished.
+ *
+ * <p>A checkpoint is triggered by sending an HTTP {@code POST} request to {@code
+ * /jobs/:jobid/checkpoints}. The HTTP request may contain a JSON body to specify a customized
+ * {@link TriggerId} and a {@link CheckpointType}, e.g.,
+ *
+ * <pre>
+ * { "triggerId": "7d273f5a62eb4730b9dea8e833733c1e", "checkpointType": "FULL" }
+ * </pre>
+ *
+ * <p>If the body is omitted, or the field {@code checkpointType} is {@code null}, the default
+ * checkpointType as specified by {@link CheckpointType#CONFIGURED} will be used. As written above,
+ * the response will contain a request id, e.g.,
+ *
+ * <pre>
+ * { "request-id": "7d273f5a62eb4730b9dea8e833733c1e" }
+ * </pre>
+ *
+ * <p>To poll for the status of an ongoing checkpoint, an HTTP {@code GET} request is issued to
+ * {@code /jobs/:jobid/checkpoints/status/:checkpointtriggerid}. If the specified checkpoint is
+ * still ongoing, the response will be
+ *
+ * <pre>
+ * {
+ * "status": {
+ * "id": "IN_PROGRESS"
+ * }
+ * }
+ * </pre>
+ *
+ * <p>If the specified checkpoints has completed, the status id will transition to {@code
+ * COMPLETED}, and the response will additionally contain information about the savepoint, such as
+ * the location:
+ *
+ * <pre>
+ * {
+ * "status": {
+ * "id": "COMPLETED"
+ * },
+ * "operation": {
+ * "checkpointId": "123"
+ * }
+ * }
+ * </pre>
+ */
+public class CheckpointTriggerHandlers {
+
+ /** handler for the checkpoint trigger operation. */
+ public static class CheckpointTriggerHandler
+ extends AbstractRestHandler<
+ RestfulGateway,
+ CheckpointTriggerRequestBody,
+ TriggerResponse,
+ CheckpointTriggerMessageParameters> {
+
+ public CheckpointTriggerHandler(
+ final GatewayRetriever<? extends RestfulGateway> leaderRetriever,
+ final Time timeout,
+ final Map<String, String> responseHeaders) {
+ super(
+ leaderRetriever,
+ timeout,
+ responseHeaders,
+ CheckpointTriggerHeaders.getInstance());
+ }
+
+ protected AsynchronousJobOperationKey createOperationKey(
+ final HandlerRequest<CheckpointTriggerRequestBody> request) {
+ final JobID jobId = request.getPathParameter(JobIDPathParameter.class);
+ return AsynchronousJobOperationKey.of(
+ request.getRequestBody().getTriggerId().orElseGet(TriggerId::new), jobId);
+ }
+
+ @Override
+ protected CompletableFuture<TriggerResponse> handleRequest(
+ @Nonnull HandlerRequest<CheckpointTriggerRequestBody> request,
+ @Nonnull RestfulGateway gateway)
+ throws RestHandlerException {
+ final AsynchronousJobOperationKey operationKey = createOperationKey(request);
+
+ return gateway.triggerCheckpoint(
+ operationKey,
+ request.getRequestBody().getCheckpointType(),
+ RpcUtils.INF_TIMEOUT)
+ .handle(
+ (acknowledge, throwable) -> {
+ if (throwable == null) {
+ return new TriggerResponse(operationKey.getTriggerId());
+ } else {
+ throw new CompletionException(
+ createInternalServerError(
+ throwable, operationKey, "triggering"));
+ }
+ });
+ }
+ }
+
+ /** HTTP handler to query for the status of the checkpoint trigger. */
+ public static class CheckpointTriggerStatusHandler
+ extends AbstractRestHandler<
+ RestfulGateway,
+ EmptyRequestBody,
+ AsynchronousOperationResult<CheckpointTriggerInfo>,
+ CheckpointTriggerStatusMessageParameters> {
+
+ public CheckpointTriggerStatusHandler(
+ final GatewayRetriever<? extends RestfulGateway> leaderRetriever,
+ final Time timeout,
+ final Map<String, String> responseHeaders) {
+ super(
+ leaderRetriever,
+ timeout,
+ responseHeaders,
+ CheckpointTriggerStatusHeaders.getInstance());
+ }
+
+ @Override
+ public CompletableFuture<AsynchronousOperationResult<CheckpointTriggerInfo>> handleRequest(
+ @Nonnull HandlerRequest<EmptyRequestBody> request, @Nonnull RestfulGateway gateway)
+ throws RestHandlerException {
+
+ final AsynchronousJobOperationKey key = getOperationKey(request);
+
+ return gateway.getTriggeredCheckpointStatus(key)
+ .handle(
+ (operationResult, throwable) -> {
+ if (throwable == null) {
+ switch (operationResult.getStatus()) {
+ case SUCCESS:
+ return AsynchronousOperationResult.completed(
+ operationResultResponse(
+ operationResult.getResult()));
+ case FAILURE:
+ return AsynchronousOperationResult.completed(
+ exceptionalOperationResultResponse(
+ operationResult.getThrowable()));
+ case IN_PROGRESS:
+ return AsynchronousOperationResult.inProgress();
+ default:
+ throw new IllegalStateException(
+ "No handler for operation status "
+ + operationResult.getStatus()
+ + ", encountered for key "
+ + key);
+ }
+ } else {
+ throw new CompletionException(
+ maybeCreateNotFoundError(throwable, key)
+ .orElseGet(
+ () ->
+ createInternalServerError(
+ throwable,
+ key,
+ "retrieving status of")));
+ }
+ });
+ }
+
+ private static Optional<RestHandlerException> maybeCreateNotFoundError(
+ Throwable throwable, AsynchronousJobOperationKey key) {
+ if (ExceptionUtils.findThrowable(throwable, UnknownOperationKeyException.class)
+ .isPresent()) {
+ return Optional.of(
+ new RestHandlerException(
+ String.format(
+ "There is no checkpoint operation with triggerId=%s for job %s.",
+ key.getTriggerId(), key.getJobId()),
+ HttpResponseStatus.NOT_FOUND));
+ }
+ return Optional.empty();
+ }
+
+ protected AsynchronousJobOperationKey getOperationKey(
+ HandlerRequest<EmptyRequestBody> request) {
+ final TriggerId triggerId = request.getPathParameter(TriggerIdPathParameter.class);
+ final JobID jobId = request.getPathParameter(JobIDPathParameter.class);
+ return AsynchronousJobOperationKey.of(triggerId, jobId);
+ }
+
+ protected CheckpointTriggerInfo exceptionalOperationResultResponse(
+ final Throwable throwable) {
+ return new CheckpointTriggerInfo(null, new SerializedThrowable(throwable));
+ }
+
+ protected CheckpointTriggerInfo operationResultResponse(final Long checkpointId) {
+ return new CheckpointTriggerInfo(checkpointId, null);
+ }
Review Comment:
```suggestion
private static CheckpointTriggerInfo operationResultResponse(final Long checkpointId) {
return new CheckpointTriggerInfo(checkpointId, null);
}
```
##########
flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/checkpoints/CheckpointTriggerHandlers.java:
##########
@@ -0,0 +1,261 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.rest.handler.job.checkpoints;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.core.execution.CheckpointType;
+import org.apache.flink.runtime.dispatcher.UnknownOperationKeyException;
+import org.apache.flink.runtime.rest.handler.AbstractRestHandler;
+import org.apache.flink.runtime.rest.handler.HandlerRequest;
+import org.apache.flink.runtime.rest.handler.RestHandlerException;
+import org.apache.flink.runtime.rest.handler.async.AsynchronousOperationResult;
+import org.apache.flink.runtime.rest.handler.async.TriggerResponse;
+import org.apache.flink.runtime.rest.handler.job.AsynchronousJobOperationKey;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.JobIDPathParameter;
+import org.apache.flink.runtime.rest.messages.TriggerId;
+import org.apache.flink.runtime.rest.messages.TriggerIdPathParameter;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerHeaders;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerInfo;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerMessageParameters;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerRequestBody;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerStatusHeaders;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerStatusMessageParameters;
+import org.apache.flink.runtime.rpc.RpcUtils;
+import org.apache.flink.runtime.webmonitor.RestfulGateway;
+import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.SerializedThrowable;
+
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+import javax.annotation.Nonnull;
+
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+
+/**
+ * HTTP handlers for asynchronous triggering of checkpoints.
+ *
+ * <p>Drawing checkpoints is a potentially long-running operation. To avoid blocking HTTP
+ * connections, checkpoints must be drawn in two steps. First, an HTTP request is issued to trigger
+ * the checkpoint asynchronously. The request will be assigned a {@link TriggerId}, which is
+ * returned in the response body. Next, the returned {@link TriggerId} should be used to poll the
+ * status of the checkpoint until it is finished.
+ *
+ * <p>A checkpoint is triggered by sending an HTTP {@code POST} request to {@code
+ * /jobs/:jobid/checkpoints}. The HTTP request may contain a JSON body to specify a customized
+ * {@link TriggerId} and a {@link CheckpointType}, e.g.,
+ *
+ * <pre>
+ * { "triggerId": "7d273f5a62eb4730b9dea8e833733c1e", "checkpointType": "FULL" }
+ * </pre>
+ *
+ * <p>If the body is omitted, or the field {@code checkpointType} is {@code null}, the default
+ * checkpointType as specified by {@link CheckpointType#CONFIGURED} will be used. As written above,
+ * the response will contain a request id, e.g.,
+ *
+ * <pre>
+ * { "request-id": "7d273f5a62eb4730b9dea8e833733c1e" }
+ * </pre>
+ *
+ * <p>To poll for the status of an ongoing checkpoint, an HTTP {@code GET} request is issued to
+ * {@code /jobs/:jobid/checkpoints/status/:checkpointtriggerid}. If the specified checkpoint is
+ * still ongoing, the response will be
+ *
+ * <pre>
+ * {
+ * "status": {
+ * "id": "IN_PROGRESS"
+ * }
+ * }
+ * </pre>
+ *
+ * <p>If the specified checkpoints has completed, the status id will transition to {@code
+ * COMPLETED}, and the response will additionally contain information about the savepoint, such as
+ * the location:
+ *
+ * <pre>
+ * {
+ * "status": {
+ * "id": "COMPLETED"
+ * },
+ * "operation": {
+ * "checkpointId": "123"
+ * }
+ * }
+ * </pre>
+ */
+public class CheckpointTriggerHandlers {
+
+ /** handler for the checkpoint trigger operation. */
+ public static class CheckpointTriggerHandler
+ extends AbstractRestHandler<
+ RestfulGateway,
+ CheckpointTriggerRequestBody,
+ TriggerResponse,
+ CheckpointTriggerMessageParameters> {
+
+ public CheckpointTriggerHandler(
+ final GatewayRetriever<? extends RestfulGateway> leaderRetriever,
+ final Time timeout,
+ final Map<String, String> responseHeaders) {
+ super(
+ leaderRetriever,
+ timeout,
+ responseHeaders,
+ CheckpointTriggerHeaders.getInstance());
+ }
+
+ protected AsynchronousJobOperationKey createOperationKey(
+ final HandlerRequest<CheckpointTriggerRequestBody> request) {
+ final JobID jobId = request.getPathParameter(JobIDPathParameter.class);
+ return AsynchronousJobOperationKey.of(
+ request.getRequestBody().getTriggerId().orElseGet(TriggerId::new), jobId);
+ }
+
+ @Override
+ protected CompletableFuture<TriggerResponse> handleRequest(
+ @Nonnull HandlerRequest<CheckpointTriggerRequestBody> request,
+ @Nonnull RestfulGateway gateway)
+ throws RestHandlerException {
+ final AsynchronousJobOperationKey operationKey = createOperationKey(request);
+
+ return gateway.triggerCheckpoint(
+ operationKey,
+ request.getRequestBody().getCheckpointType(),
+ RpcUtils.INF_TIMEOUT)
+ .handle(
+ (acknowledge, throwable) -> {
+ if (throwable == null) {
+ return new TriggerResponse(operationKey.getTriggerId());
+ } else {
+ throw new CompletionException(
+ createInternalServerError(
+ throwable, operationKey, "triggering"));
+ }
+ });
+ }
+ }
+
+ /** HTTP handler to query for the status of the checkpoint trigger. */
+ public static class CheckpointTriggerStatusHandler
+ extends AbstractRestHandler<
+ RestfulGateway,
+ EmptyRequestBody,
+ AsynchronousOperationResult<CheckpointTriggerInfo>,
+ CheckpointTriggerStatusMessageParameters> {
+
+ public CheckpointTriggerStatusHandler(
+ final GatewayRetriever<? extends RestfulGateway> leaderRetriever,
+ final Time timeout,
+ final Map<String, String> responseHeaders) {
+ super(
+ leaderRetriever,
+ timeout,
+ responseHeaders,
+ CheckpointTriggerStatusHeaders.getInstance());
+ }
+
+ @Override
+ public CompletableFuture<AsynchronousOperationResult<CheckpointTriggerInfo>> handleRequest(
+ @Nonnull HandlerRequest<EmptyRequestBody> request, @Nonnull RestfulGateway gateway)
+ throws RestHandlerException {
+
+ final AsynchronousJobOperationKey key = getOperationKey(request);
+
+ return gateway.getTriggeredCheckpointStatus(key)
+ .handle(
+ (operationResult, throwable) -> {
+ if (throwable == null) {
+ switch (operationResult.getStatus()) {
+ case SUCCESS:
+ return AsynchronousOperationResult.completed(
+ operationResultResponse(
+ operationResult.getResult()));
+ case FAILURE:
+ return AsynchronousOperationResult.completed(
+ exceptionalOperationResultResponse(
+ operationResult.getThrowable()));
+ case IN_PROGRESS:
+ return AsynchronousOperationResult.inProgress();
+ default:
+ throw new IllegalStateException(
+ "No handler for operation status "
+ + operationResult.getStatus()
+ + ", encountered for key "
+ + key);
+ }
+ } else {
+ throw new CompletionException(
+ maybeCreateNotFoundError(throwable, key)
+ .orElseGet(
+ () ->
+ createInternalServerError(
+ throwable,
+ key,
+ "retrieving status of")));
+ }
+ });
+ }
+
+ private static Optional<RestHandlerException> maybeCreateNotFoundError(
+ Throwable throwable, AsynchronousJobOperationKey key) {
+ if (ExceptionUtils.findThrowable(throwable, UnknownOperationKeyException.class)
+ .isPresent()) {
+ return Optional.of(
+ new RestHandlerException(
+ String.format(
+ "There is no checkpoint operation with triggerId=%s for job %s.",
+ key.getTriggerId(), key.getJobId()),
+ HttpResponseStatus.NOT_FOUND));
+ }
+ return Optional.empty();
+ }
+
+ protected AsynchronousJobOperationKey getOperationKey(
+ HandlerRequest<EmptyRequestBody> request) {
+ final TriggerId triggerId = request.getPathParameter(TriggerIdPathParameter.class);
+ final JobID jobId = request.getPathParameter(JobIDPathParameter.class);
+ return AsynchronousJobOperationKey.of(triggerId, jobId);
+ }
Review Comment:
```suggestion
private static AsynchronousJobOperationKey getOperationKey(
HandlerRequest<EmptyRequestBody> request) {
final TriggerId triggerId = request.getPathParameter(TriggerIdPathParameter.class);
final JobID jobId = request.getPathParameter(JobIDPathParameter.class);
return AsynchronousJobOperationKey.of(triggerId, jobId);
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] leletan commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
leletan commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r992626495
##########
flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/checkpoints/CheckpointTriggerHandlersTest.java:
##########
@@ -0,0 +1,349 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.rest.handler.job.checkpoints;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.core.execution.CheckpointType;
+import org.apache.flink.runtime.dispatcher.UnknownOperationKeyException;
+import org.apache.flink.runtime.messages.Acknowledge;
+import org.apache.flink.runtime.rest.RestMatchers;
+import org.apache.flink.runtime.rest.handler.HandlerRequest;
+import org.apache.flink.runtime.rest.handler.HandlerRequestException;
+import org.apache.flink.runtime.rest.handler.RestHandlerException;
+import org.apache.flink.runtime.rest.handler.async.AsynchronousOperationResult;
+import org.apache.flink.runtime.rest.handler.async.OperationResult;
+import org.apache.flink.runtime.rest.handler.job.AsynchronousJobOperationKey;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.JobIDPathParameter;
+import org.apache.flink.runtime.rest.messages.TriggerId;
+import org.apache.flink.runtime.rest.messages.TriggerIdPathParameter;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerInfo;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerMessageParameters;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerRequestBody;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerStatusMessageParameters;
+import org.apache.flink.runtime.rest.messages.queue.QueueStatus;
+import org.apache.flink.runtime.webmonitor.RestfulGateway;
+import org.apache.flink.runtime.webmonitor.TestingRestfulGateway;
+import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
+import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.concurrent.FutureUtils;
+
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/** Test for {@link CheckpointTriggerHandlers}. */
+public class CheckpointTriggerHandlersTest extends TestLogger {
+
+ private static final Time TIMEOUT = Time.seconds(10);
+
+ private static final JobID JOB_ID = new JobID();
+
+ private static final Long COMPLETED_CHECKPOINT_ID = 123456L;
+
+ private static CheckpointTriggerHandlers.CheckpointTriggerHandler checkpointTriggerHandler;
+
+ private static CheckpointTriggerHandlers.CheckpointTriggerStatusHandler
+ checkpointTriggerStatusHandler;
+
+ @BeforeAll
+ static void setUp() throws Exception {
+ GatewayRetriever<RestfulGateway> leaderRetriever =
+ () -> CompletableFuture.completedFuture(null);
+
+ checkpointTriggerHandler =
+ new CheckpointTriggerHandlers.CheckpointTriggerHandler(
+ leaderRetriever, TIMEOUT, Collections.emptyMap());
+
+ checkpointTriggerStatusHandler =
+ new CheckpointTriggerHandlers.CheckpointTriggerStatusHandler(
+ leaderRetriever, TIMEOUT, Collections.emptyMap());
+ }
+
+ @Test
+ public void testCheckpointTriggerCompletedSuccessfully() throws Exception {
+ final OperationResult<Long> successfulResult =
+ OperationResult.success(COMPLETED_CHECKPOINT_ID);
+ final CompletableFuture<CheckpointType> checkpointPropertiesFuture =
+ new CompletableFuture<>();
+
+ final AtomicReference<AsynchronousJobOperationKey> keyReference = new AtomicReference<>();
+ final TestingRestfulGateway testingRestfulGateway =
+ new TestingRestfulGateway.Builder()
+ .setTriggerCheckpointFunction(
+ (AsynchronousJobOperationKey key,
+ CheckpointType checkpointType) -> {
+ keyReference.set(key);
+ checkpointPropertiesFuture.complete(checkpointType);
+ return CompletableFuture.completedFuture(Acknowledge.get());
+ })
+ .setGetCheckpointStatusFunction(
+ (AsynchronousJobOperationKey operationKey) -> {
+ if (operationKey.equals(keyReference.get())) {
+ return CompletableFuture.completedFuture(successfulResult);
+ }
+ throw new RuntimeException(
+ "Expected operation key "
+ + keyReference.get()
+ + ", but received "
+ + operationKey);
+ })
+ .build();
+
+ final CheckpointType checkpointType = CheckpointType.FULL;
+
+ final TriggerId triggerId =
+ checkpointTriggerHandler
+ .handleRequest(
+ triggerCheckpointRequest(checkpointType, null),
+ testingRestfulGateway)
+ .get()
+ .getTriggerId();
+
+ final AsynchronousOperationResult<CheckpointTriggerInfo> checkpointTriggerResponseBody =
+ checkpointTriggerStatusHandler
+ .handleRequest(
+ checkpointTriggerStatusRequest(triggerId), testingRestfulGateway)
+ .get();
+
+ assertEquals(checkpointTriggerResponseBody.queueStatus().getId(), QueueStatus.Id.COMPLETED);
+ assertNotNull(checkpointTriggerResponseBody.resource());
+ assertEquals(
+ checkpointTriggerResponseBody.resource().getCheckpointId(),
+ COMPLETED_CHECKPOINT_ID);
+ assertEquals(checkpointPropertiesFuture.get(), CheckpointType.FULL);
+ }
+
+ @Test
+ public void testTriggerCheckpointNoCheckpointType() throws Exception {
+ final OperationResult<Long> successfulResult =
+ OperationResult.success(COMPLETED_CHECKPOINT_ID);
+ final CompletableFuture<CheckpointType> checkpointTypeFuture = new CompletableFuture<>();
+
+ final AtomicReference<AsynchronousJobOperationKey> keyReference = new AtomicReference<>();
+ final TestingRestfulGateway testingRestfulGateway =
+ new TestingRestfulGateway.Builder()
+ .setTriggerCheckpointFunction(
+ (AsynchronousJobOperationKey key,
+ CheckpointType checkpointType) -> {
+ keyReference.set(key);
+ checkpointTypeFuture.complete(checkpointType);
+ return CompletableFuture.completedFuture(Acknowledge.get());
+ })
+ .setGetCheckpointStatusFunction(
+ (AsynchronousJobOperationKey operationKey) -> {
+ if (operationKey.equals(keyReference.get())) {
+ return CompletableFuture.completedFuture(successfulResult);
+ }
+ throw new RuntimeException(
+ "Expected operation key "
+ + keyReference.get()
+ + ", but received "
+ + operationKey);
+ })
+ .build();
+
+ final TriggerId triggerId =
+ checkpointTriggerHandler
+ .handleRequest(triggerCheckpointRequest(null, null), testingRestfulGateway)
+ .get()
+ .getTriggerId();
+
+ AsynchronousOperationResult<CheckpointTriggerInfo> checkpointTriggerResponseBody;
+ checkpointTriggerResponseBody =
+ checkpointTriggerStatusHandler
+ .handleRequest(
+ checkpointTriggerStatusRequest(triggerId), testingRestfulGateway)
+ .get();
+
+ assertEquals(checkpointTriggerResponseBody.queueStatus().getId(), QueueStatus.Id.COMPLETED);
+ assertNotNull(checkpointTriggerResponseBody.resource());
+ assertEquals(
+ checkpointTriggerResponseBody.resource().getCheckpointId(),
+ COMPLETED_CHECKPOINT_ID);
+ assertEquals(checkpointTypeFuture.get(), CheckpointType.DEFAULT);
+ }
+
+ @Test
+ public void testCheckpointCompletedWithException() throws Exception {
+ final OperationResult<Long> failedResult =
+ OperationResult.failure(new RuntimeException("expected"));
+
+ final AtomicReference<AsynchronousJobOperationKey> keyReference = new AtomicReference<>();
+ final TestingRestfulGateway testingRestfulGateway =
+ new TestingRestfulGateway.Builder()
+ .setTriggerCheckpointFunction(
+ (AsynchronousJobOperationKey key,
+ CheckpointType checkpointType) -> {
+ keyReference.set(key);
+ return CompletableFuture.completedFuture(Acknowledge.get());
+ })
+ .setGetCheckpointStatusFunction(
+ (AsynchronousJobOperationKey operationKey) -> {
+ if (operationKey.equals(keyReference.get())) {
+ return CompletableFuture.completedFuture(failedResult);
+ }
+ throw new RuntimeException(
+ "Expected operation key "
+ + keyReference.get()
+ + ", but received "
+ + operationKey);
+ })
+ .build();
+
+ final TriggerId triggerId =
+ checkpointTriggerHandler
+ .handleRequest(triggerCheckpointRequest(null, null), testingRestfulGateway)
+ .get()
+ .getTriggerId();
+
+ AsynchronousOperationResult<CheckpointTriggerInfo> checkpointTriggerResponseBody;
+ checkpointTriggerResponseBody =
+ checkpointTriggerStatusHandler
+ .handleRequest(
+ checkpointTriggerStatusRequest(triggerId), testingRestfulGateway)
+ .get();
+
+ assertEquals(checkpointTriggerResponseBody.queueStatus().getId(), QueueStatus.Id.COMPLETED);
+ assertNotNull(checkpointTriggerResponseBody.resource());
+ assertNotNull(checkpointTriggerResponseBody.resource().getFailureCause());
+
+ final Throwable checkpointError =
+ checkpointTriggerResponseBody
+ .resource()
+ .getFailureCause()
+ .deserializeError(ClassLoader.getSystemClassLoader());
+ assertThat(checkpointError.getMessage()).matches("expected");
+ assertInstanceOf(RuntimeException.class, checkpointError);
+ }
+
+ @Test
+ public void testProvidedTriggerId() throws Exception {
+ final OperationResult<Long> successfulResult =
+ OperationResult.success(COMPLETED_CHECKPOINT_ID);
+ final AtomicReference<AsynchronousJobOperationKey> keyReference = new AtomicReference<>();
+ final TestingRestfulGateway testingRestfulGateway =
+ new TestingRestfulGateway.Builder()
+ .setTriggerCheckpointFunction(
+ (AsynchronousJobOperationKey key,
+ CheckpointType checkpointType) -> {
+ keyReference.set(key);
+ return CompletableFuture.completedFuture(Acknowledge.get());
+ })
+ .setGetCheckpointStatusFunction(
+ (AsynchronousJobOperationKey operationKey) -> {
+ if (operationKey.equals(keyReference.get())) {
+ return CompletableFuture.completedFuture(successfulResult);
+ }
+ throw new RuntimeException(
+ "Expected operation key "
+ + keyReference.get()
+ + ", but received "
+ + operationKey);
+ })
+ .build();
+
+ final TriggerId providedTriggerId = new TriggerId();
+
+ final TriggerId returnedTriggerId =
+ checkpointTriggerHandler
+ .handleRequest(
+ triggerCheckpointRequest(CheckpointType.FULL, providedTriggerId),
+ testingRestfulGateway)
+ .get()
+ .getTriggerId();
+
+ assertEquals(providedTriggerId, returnedTriggerId);
+
+ AsynchronousOperationResult<CheckpointTriggerInfo> checkpointTriggerResponseBody;
+ checkpointTriggerResponseBody =
+ checkpointTriggerStatusHandler
+ .handleRequest(
+ checkpointTriggerStatusRequest(providedTriggerId),
+ testingRestfulGateway)
+ .get();
+
+ assertEquals(checkpointTriggerResponseBody.queueStatus().getId(), QueueStatus.Id.COMPLETED);
+ assertNotNull(checkpointTriggerResponseBody.resource());
Review Comment:
I see what you mean by assertj now. Will fix.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] leletan commented on pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
leletan commented on PR #20852:
URL: https://github.com/apache/flink/pull/20852#issuecomment-1259021651
Hi folks, @pnowojski @stevenzwu @zentol I believe I addressed all comments except the pending naming of the old CheckpointType class name.
Personally I don't have any strong opinion about the class name so I just marked it as `@Internal` for now. I am happy to update it after we reach a consensus.
Meanwhile please let me know if everything else looks good to you.
Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
pnowojski commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r980923128
##########
flink-core/src/main/java/org/apache/flink/core/execution/CheckpointBackupType.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * 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.core.execution;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.DescribedEnum;
+import org.apache.flink.configuration.description.InlineElement;
+
+import static org.apache.flink.configuration.description.TextElement.text;
+
+/** Describes the backup type in which a checkpoint should be taken. */
+@PublicEvolving
+public enum CheckpointBackupType implements DescribedEnum {
Review Comment:
> `CheckpointSnapshotType` and `SavepointSnapshotType`
Either this or `Internal` infix are fine by me 👍
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] stevenzwu commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r981949061
##########
flink-core/src/main/java/org/apache/flink/core/execution/CheckpointType.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * 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.core.execution;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.DescribedEnum;
+import org.apache.flink.configuration.description.InlineElement;
+
+import static org.apache.flink.configuration.description.TextElement.text;
+
+/** Describes the type in which a checkpoint should be taken. */
+@PublicEvolving
+public enum CheckpointType implements DescribedEnum {
+ /** A checkpoint type with the entire state check pointed, common for all state backends. */
+ FULL("A checkpoint type with the entire state check pointed, common for all state backends."),
+
+ /**
+ * A checkpoint type that check points only the difference between snapshots, specific for
+ * certain state backend.
+ */
+ INCREMENTAL(
+ "A checkpoint type that check points only the difference between snapshots, specific for certain state backend.");
+ private final InlineElement description;
+ public static final CheckpointType DEFAULT = CheckpointType.FULL;
+
+ CheckpointType(String description) {
+ this.description = text(description);
+ }
+
+ @Override
+ @Internal
Review Comment:
this annotation is not necessary, 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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] leletan commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
leletan commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r974889748
##########
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointRetentionPolicy.java:
##########
@@ -24,6 +24,9 @@
@Internal
public enum CheckpointRetentionPolicy {
+ /** Full Checkpoints should be retained on cancellation and failure. */
+ FULL_RETAIN_ON_CANCELLATION,
Review Comment:
Good catch @pnowojski. This was one of the places I was not very confident with. As a hindsight, totally agree it was a bad design.
The reason why I did that was as following:
I was not sure if a `CheckpointProperty` will be valid if it is too far away from [these](https://github.com/apache/flink/blob/d8a4304b892412eab4c5c19b5deb84166943d3bb/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointProperties.java#L246) pre-configured ones, so I was thinking of creating one as close to an existing pre-configured as possible. Meanwhile the public method to create a pre-configured CheckpointProperty was through [this](https://github.com/apache/flink/blob/d8a4304b892412eab4c5c19b5deb84166943d3bb/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointProperties.java#L349) method with a `CheckpointRetentionPolicy`. And I did not notice the `@Internal` for the `CheckpointRetentionPolicy` just saw it was `public`.
Meanwhile I would love to hear your thoughts
1. What checkpoint configurations we should expose in the restful API? Are there more general use cases requesting other configurations for checkpoint manual triggering (other than ours)?
2. Or would it be better just to start with our specific use case - always only trigger a full checkpoint, i.e. reuse all `CheckpointProperty` and only replace `CheckpointType` with [`FULL_CHECKPOINT `](https://github.com/apache/flink/blob/d8a4304b892412eab4c5c19b5deb84166943d3bb/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointType.java#L30)
##########
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointRetentionPolicy.java:
##########
@@ -24,6 +24,9 @@
@Internal
public enum CheckpointRetentionPolicy {
+ /** Full Checkpoints should be retained on cancellation and failure. */
+ FULL_RETAIN_ON_CANCELLATION,
Review Comment:
Good catch @pnowojski. This was one of the places I was not very confident with. As a hindsight, totally agree it was a bad design.
The reason why I did that was as following:
I was not sure if a `CheckpointProperty` will be valid if it is too far away from [these](https://github.com/apache/flink/blob/d8a4304b892412eab4c5c19b5deb84166943d3bb/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointProperties.java#L246) pre-configured ones, so I was thinking of creating one as close to an existing pre-configured as possible. Meanwhile the public method to create a pre-configured `CheckpointProperty` was through [this](https://github.com/apache/flink/blob/d8a4304b892412eab4c5c19b5deb84166943d3bb/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointProperties.java#L349) method with a `CheckpointRetentionPolicy`. And I did not notice the `@Internal` for the `CheckpointRetentionPolicy` just saw it was `public`.
Meanwhile I would love to hear your thoughts
1. What checkpoint configurations we should expose in the restful API? Are there more general use cases requesting other configurations for checkpoint manual triggering (other than ours)?
2. Or would it be better just to start with our specific use case - always only trigger a full checkpoint, i.e. reuse all `CheckpointProperty` and only replace `CheckpointType` with [`FULL_CHECKPOINT `](https://github.com/apache/flink/blob/d8a4304b892412eab4c5c19b5deb84166943d3bb/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointType.java#L30)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] leletan commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
leletan commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r974889748
##########
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointRetentionPolicy.java:
##########
@@ -24,6 +24,9 @@
@Internal
public enum CheckpointRetentionPolicy {
+ /** Full Checkpoints should be retained on cancellation and failure. */
+ FULL_RETAIN_ON_CANCELLATION,
Review Comment:
Good catch @pnowojski. This was one of the places I was not very confident with. As a hindsight, totally agree it was a bad design.
The reason why I did that was as following:
I was not sure if a `CheckpointProperty` will be valid if it is too far away from [these](https://github.com/apache/flink/blob/d8a4304b892412eab4c5c19b5deb84166943d3bb/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointProperties.java#L246) pre-configured ones, so I was thinking of creating one as close to an existing pre-configured as possible. Meanwhile the public method to create a pre-configured `CheckpointProperty` was through [this](https://github.com/apache/flink/blob/d8a4304b892412eab4c5c19b5deb84166943d3bb/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointProperties.java#L349) method with a `CheckpointRetentionPolicy`. And I did not notice the `@Internal` for the `CheckpointRetentionPolicy` just saw it was `public`.
Meanwhile I would love to hear your thoughts
1. Would it be better just to start with our specific use case - always only trigger a full checkpoint, i.e. reuse all `CheckpointProperty` and only replace `CheckpointType` with [`FULL_CHECKPOINT `](https://github.com/apache/flink/blob/d8a4304b892412eab4c5c19b5deb84166943d3bb/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointType.java#L30), and then allow community to iterate on this?
2. Or we already know quite a few other use cases for checkpoint manual triggering so we can start with a more general API? In this case, what are the options we should expose in API? Or maybe better to start a mailing list thread / FLIP to discuss this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] leletan commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
leletan commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r974889748
##########
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointRetentionPolicy.java:
##########
@@ -24,6 +24,9 @@
@Internal
public enum CheckpointRetentionPolicy {
+ /** Full Checkpoints should be retained on cancellation and failure. */
+ FULL_RETAIN_ON_CANCELLATION,
Review Comment:
Good catch @pnowojski. This was one of the places I was not very confident with. As a hindsight, totally agree it was a bad design.
The reason why I did that was as following:
I was not sure if a `CheckpointProperty` will be valid if it is too far away from [these](https://github.com/apache/flink/blob/d8a4304b892412eab4c5c19b5deb84166943d3bb/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointProperties.java#L246) pre-configured ones, so I was thinking of creating one as close to an existing pre-configured as possible. Meanwhile the public method to create a pre-configured `CheckpointProperty` was through [this](https://github.com/apache/flink/blob/d8a4304b892412eab4c5c19b5deb84166943d3bb/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointProperties.java#L349) method with a `CheckpointRetentionPolicy`. And I did not notice the `@Internal` for the `CheckpointRetentionPolicy` just saw it was `public`.
Meanwhile I would love to hear your thoughts
1. What checkpoint configurations we should expose in the restful API? Are there more general use cases requesting other configurations for checkpoint manual triggering (other than ours)? Or maybe better to start a mailing list thread / FLIP to discuss this?
2. Or would it be better just to start with our specific use case - always only trigger a full checkpoint, i.e. reuse all `CheckpointProperty` and only replace `CheckpointType` with [`FULL_CHECKPOINT `](https://github.com/apache/flink/blob/d8a4304b892412eab4c5c19b5deb84166943d3bb/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointType.java#L30)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] leletan commented on pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
leletan commented on PR #20852:
URL: https://github.com/apache/flink/pull/20852#issuecomment-1272258337
@flinkbot run azure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] stevenzwu commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r979025202
##########
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java:
##########
@@ -511,6 +514,45 @@ public CompletableFuture<CompletedCheckpoint> triggerCheckpoint(boolean isPeriod
return triggerCheckpointFromCheckpointThread(checkpointProperties, null, isPeriodic);
}
+ /**
+ * Triggers one new checkpoint with the given checkpointBackupType. If the given
+ * checkpointBackupType is null, then it will fall back to use the CheckpointCoordinator's
+ * checkpointProperties. The return value is a future. It completes when the checkpoint
+ * triggered finishes or an error occurred.
+ *
+ * @param checkpointBackupType specifies the back up type of the checkpoint to trigger.
+ * @return a future to the completed checkpoint.
+ */
+ public CompletableFuture<CompletedCheckpoint> triggerCheckpoint(
+ @Nullable CheckpointBackupType checkpointBackupType) {
+
+ if (checkpointBackupType == null) {
+ return triggerCheckpointFromCheckpointThread(checkpointProperties, null, false);
Review Comment:
might be a little cleaner to avoid calling `trigger...` here. we can just resolve the `checkpointType` and call the `trigger...` at line 543. Also if we directly use `CheckpointType` and removed the `CheckpointBackupType`, this can be even simpler.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
pnowojski commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r977376090
##########
flink-core/src/main/java/org/apache/flink/core/execution/CheckpointBackupType.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * 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.core.execution;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.DescribedEnum;
+import org.apache.flink.configuration.description.InlineElement;
+
+import static org.apache.flink.configuration.description.TextElement.text;
+
+/** Describes the backup type in which a checkpoint should be taken. */
+@PublicEvolving
+public enum CheckpointBackupType implements DescribedEnum {
Review Comment:
Why did you select `CheckpointBackupType` as a name? I mean the `backup` part sounds strange to me. Is it because `CheckpointType` enum already exists internally? If so, either:
- I wouldn't worry too much, and having internal and public class named the same in different packages is not great, but acceptable
- or I would rename the internal one, and keep the public facing named correctly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] stevenzwu commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r979038031
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DispatcherCachedOperationsHandler.java:
##########
@@ -39,30 +40,66 @@ public class DispatcherCachedOperationsHandler {
private final CompletedOperationCache<AsynchronousJobOperationKey, String>
savepointTriggerCache;
+ private final CompletedOperationCache<AsynchronousJobOperationKey, Long> checkpointTriggerCache;
+
+ private final TriggerCheckpointFunction triggerCheckpointFunction;
+
private final TriggerSavepointFunction triggerSavepointFunction;
private final TriggerSavepointFunction stopWithSavepointFunction;
DispatcherCachedOperationsHandler(
DispatcherOperationCaches operationCaches,
+ TriggerCheckpointFunction triggerCheckpointFunction,
TriggerSavepointFunction triggerSavepointFunction,
TriggerSavepointFunction stopWithSavepointFunction) {
this(
+ triggerCheckpointFunction,
+ operationCaches.getCheckpointTriggerCache(),
triggerSavepointFunction,
stopWithSavepointFunction,
operationCaches.getSavepointTriggerCache());
}
@VisibleForTesting
DispatcherCachedOperationsHandler(
+ TriggerCheckpointFunction triggerCheckpointFunction,
+ CompletedOperationCache<AsynchronousJobOperationKey, Long> checkpointTriggerCache,
TriggerSavepointFunction triggerSavepointFunction,
TriggerSavepointFunction stopWithSavepointFunction,
CompletedOperationCache<AsynchronousJobOperationKey, String> savepointTriggerCache) {
+ this.triggerCheckpointFunction = triggerCheckpointFunction;
+ this.checkpointTriggerCache = checkpointTriggerCache;
this.triggerSavepointFunction = triggerSavepointFunction;
this.stopWithSavepointFunction = stopWithSavepointFunction;
this.savepointTriggerCache = savepointTriggerCache;
}
+ public CompletableFuture<Acknowledge> triggerCheckpoint(
+ AsynchronousJobOperationKey operationKey,
+ CheckpointBackupType checkpointBackupType,
+ Time timeout) {
+
+ if (!checkpointTriggerCache.containsOperation(operationKey)) {
+ checkpointTriggerCache.registerOngoingOperation(
+ operationKey,
+ triggerCheckpointFunction.apply(
+ operationKey.getJobId(), checkpointBackupType, timeout));
+ }
+
+ return CompletableFuture.completedFuture(Acknowledge.get());
+ }
+
+ public CompletableFuture<OperationResult<Long>> getCheckpointStatus(
Review Comment:
I got it now after seeing the other discussion. it is for checkpoint id. For savepoint trigger, the result is a path/location. for checkpoint trigger, the result is a checkpoint id.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] stevenzwu commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r978880123
##########
flink-core/src/main/java/org/apache/flink/core/execution/CheckpointBackupType.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * 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.core.execution;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.DescribedEnum;
+import org.apache.flink.configuration.description.InlineElement;
+
+import static org.apache.flink.configuration.description.TextElement.text;
+
+/** Describes the backup type in which a checkpoint should be taken. */
+@PublicEvolving
+public enum CheckpointBackupType implements DescribedEnum {
Review Comment:
@leletan @pnowojski We probably don't need this enum at all and existing public `CheckpointType` class should we sufficient. There can be two triggering modes for checkpoint via REST endpoint.
- `CHECKPOINT`: checkpoint is triggered based on the job configured checkpoint mode (incremental or full)
- `FULL_CHECKPOINT`: checkpoint is triggered for full checkpoint regardless of the job config
If in the future there is a need for triggering incremental checkpoint regardless of the job config, we can add a new enum value `INCREMENTAL_CHECKPOINT` to the existing `CheckpointType` later.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] leletan commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
leletan commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r988441596
##########
flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/checkpoints/CheckpointTriggerInfo.java:
##########
@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.rest.messages.checkpoints;
+
+import org.apache.flink.runtime.rest.messages.ResponseBody;
+import org.apache.flink.runtime.rest.messages.json.SerializedThrowableDeserializer;
+import org.apache.flink.runtime.rest.messages.json.SerializedThrowableSerializer;
+import org.apache.flink.util.SerializedThrowable;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.annotation.JsonSerialize;
+
+import javax.annotation.Nullable;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+
+/** Represents information about a finished triggered checkpoint. */
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public class CheckpointTriggerInfo implements ResponseBody {
+
+ private static final String FIELD_NAME_CHECKPOINT_ID = "checkpointId";
+
+ private static final String FIELD_NAME_FAILURE_CAUSE = "failure-cause";
Review Comment:
Was thinking of the same, but hesitated when seeing this in the [SavepointInfo](https://github.com/apache/flink/blob/99c2a415e9eeefafacf70762b6f54070f7911ceb/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/savepoints/SavepointInfo.java#L42), should we fix that as well?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] leletan commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
leletan commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r988586020
##########
flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/checkpoints/CheckpointTriggerHeaders.java:
##########
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.rest.messages.checkpoints;
+
+import org.apache.flink.runtime.rest.HttpMethodWrapper;
+import org.apache.flink.runtime.rest.handler.async.AsynchronousOperationTriggerMessageHeaders;
+import org.apache.flink.runtime.rest.messages.JobIDPathParameter;
+
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+/** These headers define the protocol for triggering a checkpoint. */
+public class CheckpointTriggerHeaders
+ extends AsynchronousOperationTriggerMessageHeaders<
+ CheckpointTriggerRequestBody, CheckpointTriggerMessageParameters> {
+
+ private static final CheckpointTriggerHeaders INSTANCE = new CheckpointTriggerHeaders();
+
+ private static final String URL =
+ String.format("/jobs/:%s/checkpoints", JobIDPathParameter.KEY);
+
+ private CheckpointTriggerHeaders() {}
+
+ @Override
+ public Class<CheckpointTriggerRequestBody> getRequestClass() {
+ return CheckpointTriggerRequestBody.class;
+ }
+
+ @Override
+ public HttpResponseStatus getResponseStatusCode() {
+ return HttpResponseStatus.ACCEPTED;
+ }
+
+ @Override
+ public CheckpointTriggerMessageParameters getUnresolvedMessageParameters() {
+ return new CheckpointTriggerMessageParameters();
+ }
+
+ @Override
+ public HttpMethodWrapper getHttpMethod() {
+ return HttpMethodWrapper.POST;
+ }
+
+ @Override
+ public String getTargetRestEndpointURL() {
+ return URL;
+ }
+
+ public static CheckpointTriggerHeaders getInstance() {
Review Comment:
will fix
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] leletan commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
leletan commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r984896981
##########
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java:
##########
@@ -511,6 +514,42 @@ public CompletableFuture<CompletedCheckpoint> triggerCheckpoint(boolean isPeriod
return triggerCheckpointFromCheckpointThread(checkpointProperties, null, isPeriodic);
}
+ /**
+ * Triggers one new checkpoint with the given checkpointType. If the given checkpointType is
+ * null, then it will fall back to use the CheckpointCoordinator's checkpointProperties. The
+ * return value is a future. It completes when the checkpoint triggered finishes or an error
+ * occurred.
+ *
+ * @param checkpointType specifies the back up type of the checkpoint to trigger.
+ * @return a future to the completed checkpoint.
+ */
+ public CompletableFuture<CompletedCheckpoint> triggerCheckpoint(
+ @Nullable CheckpointType checkpointType) {
+
+ final SnapshotType snapshotType;
+ if (checkpointType == null) {
+ snapshotType = checkpointProperties.getCheckpointType();
Review Comment:
this will be handled in a follow up PR as discussed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] leletan commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
leletan commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r980354599
##########
flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java:
##########
@@ -853,6 +855,12 @@ public CompletableFuture<ExecutionGraphInfo> requestJob(Time timeout) {
return CompletableFuture.completedFuture(schedulerNG.requestJob());
}
+ @Override
+ public CompletableFuture<CompletedCheckpoint> triggerCheckpoint(
Review Comment:
There will be no other `triggerCheckpoint` method in this class, that method is implemented in its parent interface so it will not be defined explicitly here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] leletan commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
leletan commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r992624496
##########
flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/checkpoints/CheckpointTriggerStatusHeaders.java:
##########
@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.rest.messages.checkpoints;
+
+import org.apache.flink.runtime.rest.HttpMethodWrapper;
+import org.apache.flink.runtime.rest.handler.async.AsynchronousOperationStatusMessageHeaders;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.JobIDPathParameter;
+import org.apache.flink.runtime.rest.messages.TriggerIdPathParameter;
+
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+/** These headers define the protocol for triggering a checkpoint. */
+public class CheckpointTriggerStatusHeaders
+ extends AsynchronousOperationStatusMessageHeaders<
+ CheckpointTriggerInfo, CheckpointTriggerStatusMessageParameters> {
Review Comment:
Good observation, @zentol. This is where I was struggling as well. Here are some context why I was slightly leaning towards trigger:
1. The new APIs in this PR is for manual trigger of a checkpoint and it may or may not end up with a checkpoint (it could end up with a failure and return with a failure cause, before it can reach `CheckpointCoordinator`)
2. The checkpoint trigger status is queried by trigger id and its value is read from `DispatcherOperationCaches` which has only 5 min cache duration by default, so it is more of a trigger status than checkpoint status.
3. There are already other APIs for querying checkpoint metadata and details, which should cover all the retained checkpoints for the application - don't want to confuse user with them.
Meanwhile I have to admit
1. each manual trigger only corresponds to at-most 1 checkpoint in the `DispatcherOperationCaches` thus it is making it confusing.
2. this is very similar to savepoint logic, which has no "trigger" for its API or class names.
That's why I was struggling and slightly leaning towards having "trigger" in the API and class names. However if you still feel strongly the naming should be without "trigger", I have no problem removing them.
I may missed something and feel free to let me know.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] leletan commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
leletan commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r974889748
##########
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointRetentionPolicy.java:
##########
@@ -24,6 +24,9 @@
@Internal
public enum CheckpointRetentionPolicy {
+ /** Full Checkpoints should be retained on cancellation and failure. */
+ FULL_RETAIN_ON_CANCELLATION,
Review Comment:
Good catch @pnowojski. This was one of the places I was not very confident with. As a hindsight, totally agree it was a bad design.
The reason why I did that was as following:
I was not sure if a `CheckpointProperty` will be valid if it is too far away from [these](https://github.com/apache/flink/blob/d8a4304b892412eab4c5c19b5deb84166943d3bb/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointProperties.java#L246) pre-configured ones, so I was thinking of creating one as close to an existing pre-configured as possible. Meanwhile the public method to create a pre-configured `CheckpointProperty` was through [this](https://github.com/apache/flink/blob/d8a4304b892412eab4c5c19b5deb84166943d3bb/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointProperties.java#L349) method with a `CheckpointRetentionPolicy`. And I did not notice the `@Internal` for the `CheckpointRetentionPolicy` just saw it was `public`.
Meanwhile I would love to hear your thoughts
1. Would it be better just to start with our specific use case - always only trigger a full checkpoint, i.e. reuse all `CheckpointProperty` and only replace `CheckpointType` with [`FULL_CHECKPOINT `](https://github.com/apache/flink/blob/d8a4304b892412eab4c5c19b5deb84166943d3bb/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointType.java#L30), and then allow community to iterate on this API?
2. Or we already know quite a few other use cases for checkpoint manual triggering so we can start with a more general API? In this case, what are the options we should expose in API? Or maybe better to start a mailing list thread / FLIP to discuss this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
pnowojski commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r979621391
##########
flink-core/src/main/java/org/apache/flink/core/execution/CheckpointBackupType.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * 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.core.execution;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.DescribedEnum;
+import org.apache.flink.configuration.description.InlineElement;
+
+import static org.apache.flink.configuration.description.TextElement.text;
+
+/** Describes the backup type in which a checkpoint should be taken. */
+@PublicEvolving
+public enum CheckpointBackupType implements DescribedEnum {
Review Comment:
`SnapshotTypeImpl` is not a great name choice, as the there is another implementation of `SnapshotType`.
Maybe let's rename the old `CheckpointType` and `SavepointType` to `CheckpointInternalType` and `SavepointInternalType`? 🤷
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] leletan commented on pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
leletan commented on PR #20852:
URL: https://github.com/apache/flink/pull/20852#issuecomment-1263951145
@pnowojski Your comments and change requests should all be addressed. Please let me know if I miss anything. Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] stevenzwu commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r978880123
##########
flink-core/src/main/java/org/apache/flink/core/execution/CheckpointBackupType.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * 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.core.execution;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.DescribedEnum;
+import org.apache.flink.configuration.description.InlineElement;
+
+import static org.apache.flink.configuration.description.TextElement.text;
+
+/** Describes the backup type in which a checkpoint should be taken. */
+@PublicEvolving
+public enum CheckpointBackupType implements DescribedEnum {
Review Comment:
@leletan We probably don't need this enum at all and existing `CheckpointType` should we sufficient. There can be two triggering modes for checkpoint via REST endpoint.
- `CHECKPOINT`: checkpoint is triggered based on the job configured checkpoint mode (incremental or full)
- `FULL_CHECKPOINT`: checkpoint is triggered for full checkpoint regardless of the job config
If in the future there is a need for triggering incremental checkpoint regardless of the job config, we can add a new enum value `INCREMENTAL_CHECKPOINT` to the existing `CheckpointType` later.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] stevenzwu commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r979042504
##########
flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/checkpoints/CheckpointTriggerHandlers.java:
##########
@@ -0,0 +1,261 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.rest.handler.job.checkpoints;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.runtime.dispatcher.UnknownOperationKeyException;
+import org.apache.flink.runtime.rest.handler.AbstractRestHandler;
+import org.apache.flink.runtime.rest.handler.HandlerRequest;
+import org.apache.flink.runtime.rest.handler.RestHandlerException;
+import org.apache.flink.runtime.rest.handler.async.AsynchronousOperationResult;
+import org.apache.flink.runtime.rest.handler.async.TriggerResponse;
+import org.apache.flink.runtime.rest.handler.job.AsynchronousJobOperationKey;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.JobIDPathParameter;
+import org.apache.flink.runtime.rest.messages.TriggerId;
+import org.apache.flink.runtime.rest.messages.TriggerIdPathParameter;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerHeaders;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerInfo;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerMessageParameters;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerRequestBody;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerStatusHeaders;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerStatusMessageParameters;
+import org.apache.flink.runtime.rpc.RpcUtils;
+import org.apache.flink.runtime.webmonitor.RestfulGateway;
+import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.SerializedThrowable;
+
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+import javax.annotation.Nonnull;
+
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+
+/**
+ * HTTP handlers for asynchronous triggering of checkpoints.
+ *
+ * <p>Drawing checkpoints is a potentially long-running operation. To avoid blocking HTTP
+ * connections, checkpoints must be drawn in two steps. First, an HTTP request is issued to trigger
+ * the checkpoint asynchronously. The request will be assigned a {@link TriggerId}, which is
+ * returned in the response body. Next, the returned id should be used to poll the status of the
Review Comment:
nit: `returned id` -> `returned {@link TriggerId}`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] leletan commented on pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
leletan commented on PR #20852:
URL: https://github.com/apache/flink/pull/20852#issuecomment-1251821677
> Thanks a lot for the contribution @leletan . It looks mostly good to me, I've left a couple of comments. It would be great if someone else (maybe @zentol ) could also take a look at the REST API changes.
>
> Also there seems to be a build failure.
>
> > If yes, how is the feature documented? [WIP]
>
> Let's not forget about this :)
Thanks for the super fast response, @pnowojski !
Do we have a standard convention/process about restful API feature?
- The only thing I found so far is [this](https://github.com/apache/flink/tree/master/flink-docs#rest-api-documentation) README. I am still a little bit confused after reading the README. Do we have an example for that?
- Any restful API documentation other than above we need to add?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] leletan commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
leletan commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r988585237
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java:
##########
@@ -902,6 +905,28 @@ public CompletableFuture<String> triggerCheckpoint(JobID jobID, Time timeout) {
jobID, gateway -> gateway.triggerCheckpoint(timeout));
}
+ @Override
+ public CompletableFuture<Acknowledge> triggerCheckpoint(
+ AsynchronousJobOperationKey operationKey, CheckpointType checkpointType, Time timeout) {
+ return dispatcherCachedOperationsHandler.triggerCheckpoint(
+ operationKey, checkpointType, timeout);
+ }
+
+ @Override
+ public CompletableFuture<OperationResult<Long>> getTriggeredCheckpointStatus(
+ AsynchronousJobOperationKey operationKey) {
+ return dispatcherCachedOperationsHandler.getCheckpointStatus(operationKey);
+ }
+
+ public CompletableFuture<Long> triggerCheckpointAndGetCheckpointID(
Review Comment:
Good catch. Was just following the [corresponding savepoint function](https://github.com/apache/flink/blob/9152af41c2d401e5eacddee1bb10d1b6bea6c61a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java#L916-L931). Just realized this could be unnecessary since this function is not needed in [EmbeddedJobClient](https://github.com/apache/flink/blob/cd8ea8d5b207569f68acc5a3c8db95cd2ca47ba6/flink-clients/src/main/java/org/apache/flink/client/deployment/application/EmbeddedJobClient.java)
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java:
##########
@@ -902,6 +905,28 @@ public CompletableFuture<String> triggerCheckpoint(JobID jobID, Time timeout) {
jobID, gateway -> gateway.triggerCheckpoint(timeout));
}
+ @Override
+ public CompletableFuture<Acknowledge> triggerCheckpoint(
+ AsynchronousJobOperationKey operationKey, CheckpointType checkpointType, Time timeout) {
+ return dispatcherCachedOperationsHandler.triggerCheckpoint(
+ operationKey, checkpointType, timeout);
+ }
+
+ @Override
+ public CompletableFuture<OperationResult<Long>> getTriggeredCheckpointStatus(
+ AsynchronousJobOperationKey operationKey) {
+ return dispatcherCachedOperationsHandler.getCheckpointStatus(operationKey);
+ }
+
+ public CompletableFuture<Long> triggerCheckpointAndGetCheckpointID(
Review Comment:
Will make it private
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zentol commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
zentol commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r988930245
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerNG.java:
##########
@@ -125,7 +127,8 @@ void notifyKvStateUnregistered(
CompletableFuture<String> triggerSavepoint(
@Nullable String targetDirectory, boolean cancelJob, SavepointFormatType formatType);
- CompletableFuture<String> triggerCheckpoint();
+ CompletableFuture<CompletedCheckpoint> triggerCheckpoint(
+ @Nullable CheckpointType checkpointType);
Review Comment:
Why cant we have the MiniCluster default to `CheckpointType#DEFAULT`, or whatever was set as the default based on the configuration.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] leletan commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
leletan commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r981864310
##########
flink-core/src/main/java/org/apache/flink/core/execution/CheckpointBackupType.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * 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.core.execution;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.DescribedEnum;
+import org.apache.flink.configuration.description.InlineElement;
+
+import static org.apache.flink.configuration.description.TextElement.text;
+
+/** Describes the backup type in which a checkpoint should be taken. */
+@PublicEvolving
+public enum CheckpointBackupType implements DescribedEnum {
Review Comment:
Renamed `CheckpointType` -> `CheckpointSnapshotType` and `SavepointType` -> `SavepointSnapshotType`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zentol merged pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
zentol merged PR #20852:
URL: https://github.com/apache/flink/pull/20852
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] leletan commented on pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
leletan commented on PR #20852:
URL: https://github.com/apache/flink/pull/20852#issuecomment-1273650082
Hi folks, @pnowojski @stevenzwu @zentol Thanks for your review again. I believe I addressed all comments, except the pending naming of the old CheckpointType class name (will add in a later commit to make the code easier)
One big change is, I added one more CheckpointType as `CONFIGURED`, to represent using the existing internal CheckpointSnapshotType as configured by flink application to triggerCheckpoint, replacing previous legacy minicluster case represented by `null`.
Please let me know if there is anything else I am missing.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a diff in pull request #20852: feat(flink-runtime): add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
pnowojski commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r974198871
##########
flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterGateway.java:
##########
@@ -218,13 +220,27 @@ CompletableFuture<String> triggerSavepoint(
final SavepointFormatType formatType,
@RpcTimeout final Time timeout);
+ /**
+ * Triggers taking a checkpoint of the executed job.
+ *
+ * @param checkpointProperties to determine how checkpoint should be taken or null if the
+ * existing checkpoint property should be used
+ * @param timeout for the rpc call
+ * @return Future which is completed with the savepoint path once completed
Review Comment:
nit: java doc should say it returns `{@link CompletedCheckpoint}` not `path`.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterGateway.java:
##########
@@ -218,13 +220,27 @@ CompletableFuture<String> triggerSavepoint(
final SavepointFormatType formatType,
@RpcTimeout final Time timeout);
+ /**
+ * Triggers taking a checkpoint of the executed job.
+ *
+ * @param checkpointProperties to determine how checkpoint should be taken or null if the
+ * existing checkpoint property should be used
+ * @param timeout for the rpc call
+ * @return Future which is completed with the savepoint path once completed
+ */
+ CompletableFuture<CompletedCheckpoint> triggerCheckpoint(
Review Comment:
Does it have to return `CompletedCheckpoint`? It makes a bit inconsistent with the other method. Can not we use checkpoint location as the cache key?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointRetentionPolicy.java:
##########
@@ -24,6 +24,9 @@
@Internal
public enum CheckpointRetentionPolicy {
+ /** Full Checkpoints should be retained on cancellation and failure. */
+ FULL_RETAIN_ON_CANCELLATION,
Review Comment:
Whether the checkpoint is full or not doesn't fit with the `CheckpointRetentionPolicy`. It should be an independent axis.
If the checkpoint is full or incremental it should be parametrised independently of the retention policy. Assuming that we want to expose retention to users - and I don't see why we should be doing this. Is there some particular reason why have you implemented it like this @leletan ? Such exposure makes this more complicated for us to maintain code in the future and it also makes it more complicated for the users. To me it sounds like this retention policy should come from the `CheckpointCoordinatorConfiguration` and be the same for manually and automatically triggered checkpoints.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java:
##########
@@ -511,6 +511,20 @@ public CompletableFuture<CompletedCheckpoint> triggerCheckpoint(boolean isPeriod
return triggerCheckpointFromCheckpointThread(checkpointProperties, null, isPeriodic);
}
+ /**
+ * Triggers one new checkpoint with the given checkpoint properties. If the given checkpoint
+ * properties is null, then it will fall back to use the CheckpointCoordinator's
+ * checkpointProperties. The return value is a future. It completes when the checkpoint
+ * triggered finishes or an error occurred.
+ *
+ * @param props specifies the properties of the checkpoint to trigger.
+ * @return a future to the completed checkpoint.
+ */
+ public CompletableFuture<CompletedCheckpoint> triggerCheckpoint(CheckpointProperties props) {
Review Comment:
nit: add `@Nullable` annotation in missing places
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zentol commented on pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
zentol commented on PR #20852:
URL: https://github.com/apache/flink/pull/20852#issuecomment-1251977203
re documentation: You basically just run `mvn package -Dgenerate-rest-docs -pl flink-docs -am -nsu -DskipTests` locally.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
pnowojski commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r975199952
##########
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointRetentionPolicy.java:
##########
@@ -24,6 +24,9 @@
@Internal
public enum CheckpointRetentionPolicy {
+ /** Full Checkpoints should be retained on cancellation and failure. */
+ FULL_RETAIN_ON_CANCELLATION,
Review Comment:
I could see use cases for manually triggering incremental checkpoints as well, so I would suggest to full/incremental parameter. We would need to decouple REST API constants from the internal classes and add a translation layer for that. It should be probably something similar to `org.apache.flink.core.execution.SavepointFormatType`. `CheckpointType` with two supported values `INCREMENTAL` and `FULL`.
As for the `CheckpointProperties` class, probably you should copy instead of passing `CheckpointProperties` in `org.apache.flink.runtime.checkpoint.CheckpointCoordinator#triggerCheckpoint(org.apache.flink.runtime.checkpoint.CheckpointProperties)`, you should just pass `org.apache.flink.runtime.checkpoint.CheckpointType` and inside the `triggerCheckpoint` construct new `CheckpointProperties` by copying `org.apache.flink.runtime.checkpoint.CheckpointCoordinator#checkpointProperties`. Something like that:
```
public CompletableFuture<CompletedCheckpoint> triggerCheckpoint(CheckpointType checkpointType) {
CheckpointProperties properties =
new CheckpointProperties(
checkpointProperties.forceCheckpoint(),
checkpointType,
checkpointProperties.discardOnSubsumed(),
checkpointProperties.discardOnJobFinished(),
checkpointProperties.discardOnJobCancelled(),
checkpointProperties.discardOnJobFailed(),
checkpointProperties.discardOnJobSuspended(),
checkpointProperties.isUnclaimed());
return triggerCheckpointFromCheckpointThread(properties, null, false);
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] leletan commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
leletan commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r979107588
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerBase.java:
##########
@@ -878,7 +879,8 @@ public CompletableFuture<String> triggerSavepoint(
}
@Override
- public CompletableFuture<String> triggerCheckpoint() {
+ public CompletableFuture<CompletedCheckpoint> triggerCheckpoint(
Review Comment:
Hmm, good question ... I was just following the same function call path as triggerSavepoint (resful -> dispatcher -> job master -> scheduler -> checkpoint coordinator).
@stevenzwu Are you suggesting we skip scheduler and call CheckpointCoordinator::triggerCheckpoint(...) directly from job 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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] leletan commented on pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
leletan commented on PR #20852:
URL: https://github.com/apache/flink/pull/20852#issuecomment-1254461989
@pnowojski @zentol Addressed the code comments (at least I thought I did) and generated the docs. Please take a look whenever you have time. Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zentol commented on pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
zentol commented on PR #20852:
URL: https://github.com/apache/flink/pull/20852#issuecomment-1254920557
`API was modified in a compatible way, but the snapshot was not updated. To update the snapshot, re-run this test with -Dgenerate-rest-snapshot being set.`
Run "mvn verify -Dgenerate-rest-snapshot -pl flink-runtime-web -nsu", or `RuntimeRestAPIStabilityTest` with the property being set.
Essentially there are 2 files we generate: the documentation (with my last comment) and a snapshot for comparing compatibility (we cant use the docs because we don't persist everything we need IIRC).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] stevenzwu commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r979032943
##########
flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java:
##########
@@ -863,11 +871,6 @@ public CompletableFuture<String> triggerSavepoint(
return schedulerNG.triggerSavepoint(targetDirectory, cancelJob, formatType);
}
- @Override
Review Comment:
we shouldn't remove a public method from a public class. it is also necessary here. it can be kept for backward compatibility.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] leletan commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
leletan commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r974889748
##########
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointRetentionPolicy.java:
##########
@@ -24,6 +24,9 @@
@Internal
public enum CheckpointRetentionPolicy {
+ /** Full Checkpoints should be retained on cancellation and failure. */
+ FULL_RETAIN_ON_CANCELLATION,
Review Comment:
Good catch @pnowojski. This was one of the places I was not very confident with. As a hindsight, totally agree it was a bad design.
The reason why I did that was as following:
I was not sure if a `CheckpointProperty` will be valid if it is too far away from [these](https://github.com/apache/flink/blob/d8a4304b892412eab4c5c19b5deb84166943d3bb/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointProperties.java#L246) pre-configured ones, so I was thinking of creating one as close to an existing pre-configured as possible. Meanwhile the public method to create a pre-configured `CheckpointProperty` was through [this](https://github.com/apache/flink/blob/d8a4304b892412eab4c5c19b5deb84166943d3bb/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointProperties.java#L349) method with a `CheckpointRetentionPolicy`. And I did not notice the `@Internal` for the `CheckpointRetentionPolicy` just saw it was `public`.
Meanwhile I would love to hear your thoughts
1. Would it be better just to start with our specific use case - always only trigger a full checkpoint, i.e. reuse all `CheckpointProperty` and only replace `CheckpointType` with [`FULL_CHECKPOINT `](https://github.com/apache/flink/blob/d8a4304b892412eab4c5c19b5deb84166943d3bb/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointType.java#L30), and then allow community to iterate on this API?
2. Or we already know quite a few other use cases for checkpoint manual triggering so we can start with a more general API? In this case, what are the options we should expose in this API? Or maybe better to start a mailing list thread / FLIP to discuss this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] leletan commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
leletan commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r974870165
##########
flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterGateway.java:
##########
@@ -218,13 +220,27 @@ CompletableFuture<String> triggerSavepoint(
final SavepointFormatType formatType,
@RpcTimeout final Time timeout);
+ /**
+ * Triggers taking a checkpoint of the executed job.
+ *
+ * @param checkpointProperties to determine how checkpoint should be taken or null if the
+ * existing checkpoint property should be used
+ * @param timeout for the rpc call
+ * @return Future which is completed with the savepoint path once completed
+ */
+ CompletableFuture<CompletedCheckpoint> triggerCheckpoint(
Review Comment:
Good question, @pnowojski.
Assuming we are talking about
1. `CompletableFuture<String> triggerCheckpoint(@RpcTimeout final Time timeout)` as the other method
2. `CompletedOperationCache<AsynchronousJobOperationKey, Long> checkpointTriggerCache` in the `DispatcherCachedOperationsHandler` as the cache (location as cache value?)
We definitely can. Meanwhile I guess this means that the restful API (checkpoint trigger status API) will eventually return checkpoint location in the restful response body, instead of checkpoint id.
The reason for returning `CompletedCheckpoint` was to facilitate both old method and new API: the new restful API was returning checkpoint id, while the other method was returning checkpoint location, and CompletedCheckpoint contains both.
I am pretty new to flink so I am not sure what people typically care about a checkpoint.
If we think it is a better design to return checkpoint location, instead of checkpoint id in the response body in the restful API, I am happy to return checkpoint location and reuse existing cache and another method.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
pnowojski commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r980923128
##########
flink-core/src/main/java/org/apache/flink/core/execution/CheckpointBackupType.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * 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.core.execution;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.DescribedEnum;
+import org.apache.flink.configuration.description.InlineElement;
+
+import static org.apache.flink.configuration.description.TextElement.text;
+
+/** Describes the backup type in which a checkpoint should be taken. */
+@PublicEvolving
+public enum CheckpointBackupType implements DescribedEnum {
Review Comment:
> `CheckpointSnapshotType` and `SavepointSnapshotType`
Either this or `Internal` infix, or keeping the original name (with two `CechkpointType` enums in different packages) are fine by me 👍 `CheckpointSnapshotType` and `SavepointSnapshotType` sounds best from those three proposals for me, but I'm fine with either one of them.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] leletan commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
leletan commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r980350376
##########
flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java:
##########
@@ -863,11 +871,6 @@ public CompletableFuture<String> triggerSavepoint(
return schedulerNG.triggerSavepoint(targetDirectory, cancelJob, formatType);
}
- @Override
Review Comment:
The backward compatibility should already been kept. At this class's parent interface `JobMasterGateway`, this method is supported as
```
/**
* Triggers taking a checkpoint of the executed job.
*
* @param timeout for the rpc call
* @return Future which is completed with the checkpoint path once completed
*/
default CompletableFuture<String> triggerCheckpoint(@RpcTimeout final Time timeout) {
return triggerCheckpoint(null, timeout).thenApply(CompletedCheckpoint::getExternalPointer);
}
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] leletan commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
leletan commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r988631290
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerNG.java:
##########
@@ -125,7 +127,8 @@ void notifyKvStateUnregistered(
CompletableFuture<String> triggerSavepoint(
@Nullable String targetDirectory, boolean cancelJob, SavepointFormatType formatType);
- CompletableFuture<String> triggerCheckpoint();
+ CompletableFuture<CompletedCheckpoint> triggerCheckpoint(
+ @Nullable CheckpointType checkpointType);
Review Comment:
Good question. Correct, this is nullable here but not nullable in the request body. The reason is a little bit complicated to explain:
At essence, the null is only to take care of [this](https://github.com/apache/flink/pull/17278) special case of legacy
`CompletableFuture<String> triggerCheckpoint(@RpcTimeout final Time timeout)` function.
Here are some more context:
The call sequence of the above mentioned legacy function is
**_Minicluster -> DispatcherGateway -> Dispatcher -> JobMasterGateway -> ..._**
and it expect a return type of `CompletableFuture<String>` (String for checkpointLocation). This one does not have any CheckpointType in the function parameter.
The call sequence of the new API is
**_CheckpointTriggerHandler -> RestfulGateway -> Dispatcher -> JobMasterGateway -> ..._**
and it expect a return type of `CompletableFuture<Long>` (Long for checkpointId). In this call sequence we expect `checkpointType` to be `nonNull`
To avoid keeping 2 very similar functions down the same call sequence (**_...-> JobMasterGateway -> JobMaster -> SchedulerNG -> Scheduler -> ..._**) and duplicated a lot of code, we combine them through
```
default CompletableFuture<String> triggerCheckpoint(@RpcTimeout final Time timeout) {
return triggerCheckpoint(null, timeout).thenApply(CompletedCheckpoint::getExternalPointer);
}
```
in JobMasterGateway.
We may also combine them in `Dispatcher` - the first class where 2 call sequence meet, however the legacy call sequence already had a non HA safe implementation (more context [here](https://issues.apache.org/jira/browse/FLINK-18312)) which is good enough for Minicluster but not good enough for RestfulAPI.
Please let me know if you have any questions about above.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] leletan commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
leletan commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r993738316
##########
flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/checkpoints/CheckpointTriggerStatusHeaders.java:
##########
@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.rest.messages.checkpoints;
+
+import org.apache.flink.runtime.rest.HttpMethodWrapper;
+import org.apache.flink.runtime.rest.handler.async.AsynchronousOperationStatusMessageHeaders;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.JobIDPathParameter;
+import org.apache.flink.runtime.rest.messages.TriggerIdPathParameter;
+
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+/** These headers define the protocol for triggering a checkpoint. */
+public class CheckpointTriggerStatusHeaders
+ extends AsynchronousOperationStatusMessageHeaders<
+ CheckpointTriggerInfo, CheckpointTriggerStatusMessageParameters> {
Review Comment:
I see. Will refactor to match savepoint naming.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] leletan commented on pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
leletan commented on PR #20852:
URL: https://github.com/apache/flink/pull/20852#issuecomment-1276935626
@zentol Thanks for the detailed feedbacks, really appreciated them!
I should have addressed all the comments in this round. Please take another look when you have a chance.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] leletan commented on pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
leletan commented on PR #20852:
URL: https://github.com/apache/flink/pull/20852#issuecomment-1275067537
@zentol I see I have a lot of style issues and thanks for pointing them out. I am wondering if you have a doc / pointer for those style rules so I won't make those mistakes in the future.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #20852: feat(flink-runtime): add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
flinkbot commented on PR #20852:
URL: https://github.com/apache/flink/pull/20852#issuecomment-1250666839
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7c84a40a9c328cb8d9af188fabb0995d88dad189",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "7c84a40a9c328cb8d9af188fabb0995d88dad189",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 7c84a40a9c328cb8d9af188fabb0995d88dad189 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
pnowojski commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r975206126
##########
flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterGateway.java:
##########
@@ -218,13 +220,27 @@ CompletableFuture<String> triggerSavepoint(
final SavepointFormatType formatType,
@RpcTimeout final Time timeout);
+ /**
+ * Triggers taking a checkpoint of the executed job.
+ *
+ * @param checkpointProperties to determine how checkpoint should be taken or null if the
+ * existing checkpoint property should be used
+ * @param timeout for the rpc call
+ * @return Future which is completed with the savepoint path once completed
+ */
+ CompletableFuture<CompletedCheckpoint> triggerCheckpoint(
Review Comment:
Good point. Indeed it would be better to not expose checkpoint location, but just the checkpoint id. Checkpoints are owned by the system (contrary to the savepoints, which are owned by the user), so user shouldn't be tempted to touch checkpoint files. In this case, let's keep `CompletedCheckpoint`
##########
flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterGateway.java:
##########
@@ -218,13 +220,27 @@ CompletableFuture<String> triggerSavepoint(
final SavepointFormatType formatType,
@RpcTimeout final Time timeout);
+ /**
+ * Triggers taking a checkpoint of the executed job.
+ *
+ * @param checkpointProperties to determine how checkpoint should be taken or null if the
+ * existing checkpoint property should be used
+ * @param timeout for the rpc call
+ * @return Future which is completed with the savepoint path once completed
+ */
+ CompletableFuture<CompletedCheckpoint> triggerCheckpoint(
Review Comment:
Good point. Indeed it would be better to not expose checkpoint location, but just the checkpoint id. Checkpoints are owned by the system (contrary to the savepoints, which are owned by the user), so user shouldn't be tempted to touch checkpoint files. In this case, let's keep `CompletedCheckpoint` as the return type.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] leletan commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
leletan commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r988631290
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerNG.java:
##########
@@ -125,7 +127,8 @@ void notifyKvStateUnregistered(
CompletableFuture<String> triggerSavepoint(
@Nullable String targetDirectory, boolean cancelJob, SavepointFormatType formatType);
- CompletableFuture<String> triggerCheckpoint();
+ CompletableFuture<CompletedCheckpoint> triggerCheckpoint(
+ @Nullable CheckpointType checkpointType);
Review Comment:
Good question. Correct, this is nullable here but not nullable in the request body. The reason is a little bit complicated to explain:
At essence, the `nonNull` is expected for normal restful API call all the way to CheckpointCoordinator, and the `nullable` is only to take care of [this](https://github.com/apache/flink/pull/17278) legacy case.
Here are some more context:
The call sequence of the above mentioned legacy case `CompletableFuture<String> triggerCheckpoint(@RpcTimeout final Time timeout)` is
**_Minicluster -> DispatcherGateway -> Dispatcher -> JobMasterGateway -> ..._**
and it expect a return type of `CompletableFuture<String>` (String for checkpointLocation). This one does not have any CheckpointType in the function parameter.
The call sequence of the new API is
**_CheckpointTriggerHandler -> RestfulGateway -> Dispatcher -> JobMasterGateway -> ..._**
and it expect a return type of `CompletableFuture<Long>` (Long for checkpointId). In this call sequence we expect `checkpointType` to be `nonNull`
To avoid keeping 2 very similar functions down the same call sequence (**_...-> JobMasterGateway -> JobMaster -> SchedulerNG -> Scheduler -> ..._**) and duplicated a lot of code, we combine them through
```
default CompletableFuture<String> triggerCheckpoint(@RpcTimeout final Time timeout) {
return triggerCheckpoint(null, timeout).thenApply(CompletedCheckpoint::getExternalPointer);
}
```
in JobMasterGateway.
We may also combine them in `Dispatcher` - the first class where 2 call sequence meet, however the legacy call sequence already had a non HA safe implementation (more context [here](https://issues.apache.org/jira/browse/FLINK-18312)) which is good enough for Minicluster but not good enough for RestfulAPI.
I know this is NOT a perfect solution, but it is the best I can think of after considering trade-offs among all the solutions I can find. Please let me know if you have a better 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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zentol commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
zentol commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r988926581
##########
flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/checkpoints/CheckpointTriggerInfo.java:
##########
@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.rest.messages.checkpoints;
+
+import org.apache.flink.runtime.rest.messages.ResponseBody;
+import org.apache.flink.runtime.rest.messages.json.SerializedThrowableDeserializer;
+import org.apache.flink.runtime.rest.messages.json.SerializedThrowableSerializer;
+import org.apache.flink.util.SerializedThrowable;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.annotation.JsonSerialize;
+
+import javax.annotation.Nullable;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+
+/** Represents information about a finished triggered checkpoint. */
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public class CheckpointTriggerInfo implements ResponseBody {
+
+ private static final String FIELD_NAME_CHECKPOINT_ID = "checkpointId";
+
+ private static final String FIELD_NAME_FAILURE_CAUSE = "failure-cause";
+
+ @JsonProperty(FIELD_NAME_CHECKPOINT_ID)
+ @Nullable
+ private final Long checkpointId;
+
+ @JsonProperty(FIELD_NAME_FAILURE_CAUSE)
+ @JsonSerialize(using = SerializedThrowableSerializer.class)
+ @JsonDeserialize(using = SerializedThrowableDeserializer.class)
+ @Nullable
+ private final SerializedThrowable failureCause;
+
+ @JsonCreator
+ public CheckpointTriggerInfo(
+ @JsonProperty(FIELD_NAME_CHECKPOINT_ID) @Nullable final Long checkpointId,
+ @JsonProperty(FIELD_NAME_FAILURE_CAUSE)
+ @JsonDeserialize(using = SerializedThrowableDeserializer.class)
+ @Nullable
+ final SerializedThrowable failureCause) {
+ checkArgument(
+ checkpointId != null ^ failureCause != null,
+ "Either checkpointId or failureCause must be set");
+
+ this.checkpointId = checkpointId;
+ this.failureCause = failureCause;
+ }
+
+ @Nullable
+ public Long getCheckpointId() {
+ return checkpointId;
+ }
+
+ @Nullable
+ public SerializedThrowable getFailureCause() {
+ return failureCause;
+ }
Review Comment:
You can add a hotfix commit for that, yes.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DispatcherOperationCaches.java:
##########
@@ -40,13 +42,19 @@ public DispatcherOperationCaches() {
@VisibleForTesting
public DispatcherOperationCaches(Duration cacheDuration) {
savepointTriggerCache = new CompletedOperationCache<>(cacheDuration);
+ checkpointTriggerCache = new CompletedOperationCache<>(cacheDuration);
}
public CompletedOperationCache<AsynchronousJobOperationKey, String> getSavepointTriggerCache() {
return savepointTriggerCache;
}
+ public CompletedOperationCache<AsynchronousJobOperationKey, Long> getCheckpointTriggerCache() {
+ return checkpointTriggerCache;
+ }
+
public CompletableFuture<Void> shutdownCaches() {
- return savepointTriggerCache.closeAsync();
+ return CompletableFuture.allOf(
Review Comment:
FutureUtils#completeAll is preferable because it doesn't drop exceptions should multiple futures fail.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zentol commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
zentol commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r988926165
##########
flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/checkpoints/CheckpointTriggerInfo.java:
##########
@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.rest.messages.checkpoints;
+
+import org.apache.flink.runtime.rest.messages.ResponseBody;
+import org.apache.flink.runtime.rest.messages.json.SerializedThrowableDeserializer;
+import org.apache.flink.runtime.rest.messages.json.SerializedThrowableSerializer;
+import org.apache.flink.util.SerializedThrowable;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.annotation.JsonSerialize;
+
+import javax.annotation.Nullable;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+
+/** Represents information about a finished triggered checkpoint. */
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public class CheckpointTriggerInfo implements ResponseBody {
+
+ private static final String FIELD_NAME_CHECKPOINT_ID = "checkpointId";
+
+ private static final String FIELD_NAME_FAILURE_CAUSE = "failure-cause";
Review Comment:
> should we fix that as well?
No; we defer changes to other fields for 2.0 unless there's a good reason to change 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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zentol commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
zentol commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r988930245
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerNG.java:
##########
@@ -125,7 +127,8 @@ void notifyKvStateUnregistered(
CompletableFuture<String> triggerSavepoint(
@Nullable String targetDirectory, boolean cancelJob, SavepointFormatType formatType);
- CompletableFuture<String> triggerCheckpoint();
+ CompletableFuture<CompletedCheckpoint> triggerCheckpoint(
+ @Nullable CheckpointType checkpointType);
Review Comment:
Why cant we have the MiniCluster default to `CheckpointType#DEFAULT`, or whatever was set as the default based on the configuration?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zentol commented on pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
zentol commented on PR #20852:
URL: https://github.com/apache/flink/pull/20852#issuecomment-1277917319
@flinkbot run azure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
pnowojski commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r979065292
##########
flink-core/src/main/java/org/apache/flink/core/execution/CheckpointBackupType.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * 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.core.execution;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.DescribedEnum;
+import org.apache.flink.configuration.description.InlineElement;
+
+import static org.apache.flink.configuration.description.TextElement.text;
+
+/** Describes the backup type in which a checkpoint should be taken. */
+@PublicEvolving
+public enum CheckpointBackupType implements DescribedEnum {
Review Comment:
Already existing `org.apache.flink.runtime.checkpoint.CheckpointType` is an internal class, not a public one and I asked to not expose it the users. Note for example `org.apache.flink.runtime.checkpoint.CheckpointType#getSharingFilesStrategy`, which shouldn't be exposed.
https://github.com/apache/flink/pull/20852#discussion_r975199952
Hence I suggested to create this public facing one.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] leletan commented on pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
leletan commented on PR #20852:
URL: https://github.com/apache/flink/pull/20852#issuecomment-1260284951
Thanks again folks. All comments should be address, please let me know if there is anything else I need to do for this PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] stevenzwu commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r981948068
##########
flink-core/src/main/java/org/apache/flink/core/execution/CheckpointType.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * 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.core.execution;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.DescribedEnum;
+import org.apache.flink.configuration.description.InlineElement;
+
+import static org.apache.flink.configuration.description.TextElement.text;
+
+/** Describes the type in which a checkpoint should be taken. */
+@PublicEvolving
+public enum CheckpointType implements DescribedEnum {
+ /** A checkpoint type with the entire state check pointed, common for all state backends. */
+ FULL("A checkpoint type with the entire state check pointed, common for all state backends."),
+
+ /**
+ * A checkpoint type that check points only the difference between snapshots, specific for
Review Comment:
should be one word: check points -> checkpoints.
Also I see the style difference in the description btw full and incremental. We probably can reword the description for full as following.
```
A checkpoint type that checkpoints the entire state, ...
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] stevenzwu commented on pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on PR #20852:
URL: https://github.com/apache/flink/pull/20852#issuecomment-1260402519
@leletan I think we probably should make the `CheckpointType` and `SavepointType` renaming as a separate PR. merge that one first then come back on this one. it gets difficult to review the diff when the two changes are mixed together. touching so many files.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] leletan commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
leletan commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r979107588
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerBase.java:
##########
@@ -878,7 +879,8 @@ public CompletableFuture<String> triggerSavepoint(
}
@Override
- public CompletableFuture<String> triggerCheckpoint() {
+ public CompletableFuture<CompletedCheckpoint> triggerCheckpoint(
Review Comment:
Hmm, good question ... I was just following the same function call path as triggerSavepoint (restful -> dispatcher -> job master -> scheduler -> checkpoint coordinator).
@stevenzwu Are you suggesting we skip scheduler and call CheckpointCoordinator::triggerCheckpoint(...) directly from job master?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerBase.java:
##########
@@ -878,7 +879,8 @@ public CompletableFuture<String> triggerSavepoint(
}
@Override
- public CompletableFuture<String> triggerCheckpoint() {
+ public CompletableFuture<CompletedCheckpoint> triggerCheckpoint(
Review Comment:
Hmm, good question ... I was just following the same function call path as triggerSavepoint (rest -> dispatcher -> job master -> scheduler -> checkpoint coordinator).
@stevenzwu Are you suggesting we skip scheduler and call CheckpointCoordinator::triggerCheckpoint(...) directly from job 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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] stevenzwu commented on a diff in pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on code in PR #20852:
URL: https://github.com/apache/flink/pull/20852#discussion_r980320317
##########
flink-core/src/main/java/org/apache/flink/core/execution/CheckpointBackupType.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * 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.core.execution;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.DescribedEnum;
+import org.apache.flink.configuration.description.InlineElement;
+
+import static org.apache.flink.configuration.description.TextElement.text;
+
+/** Describes the backup type in which a checkpoint should be taken. */
+@PublicEvolving
+public enum CheckpointBackupType implements DescribedEnum {
Review Comment:
agree that `SnapshotTypeImpl` is not a good choice. `Internal` would be ok for me.
Just to brainstorm more options. what about `CheckpointSnapshotType` and `SavepointSnapshotType`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] leletan commented on pull request #20852: [FLINK-27101][checkpointing][rest] Add restful API to trigger checkpoints
Posted by GitBox <gi...@apache.org>.
leletan commented on PR #20852:
URL: https://github.com/apache/flink/pull/20852#issuecomment-1263949515
@stevenzwu All your comments should be addressed. Please let me know if there is anything missing. Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org