masteryhx commented on code in PR #23253:
URL: https://github.com/apache/flink/pull/23253#discussion_r1415222035
##########
docs/content.zh/docs/deployment/cli.md:
##########
@@ -125,6 +125,43 @@ Lastly, you can optionally provide what should be the
[binary format]({{< ref "d
The path to the savepoint can be used later on to [restart the Flink
job](#starting-a-job-from-a-savepoint).
+If the state of the job is quite big, the client will get a timeout exception
since it should wait for the savepoint finished.
Review Comment:
```suggestion
If the state size of the job is quite big, the client will get a timeout
exception since it has to wait for the savepoint finished.
```
##########
docs/content.zh/docs/deployment/cli.md:
##########
@@ -125,6 +125,43 @@ Lastly, you can optionally provide what should be the
[binary format]({{< ref "d
The path to the savepoint can be used later on to [restart the Flink
job](#starting-a-job-from-a-savepoint).
+If the state of the job is quite big, the client will get a timeout exception
since it should wait for the savepoint finished.
+```
+Triggering savepoint for job bec5244e09634ad71a80785937a9732d.
+Waiting for response...
+
+--------------------------------------------------------------
+The program finished with the following exception:
+
+org.apache.flink.util.FlinkException: Triggering a savepoint for the job
bec5244e09634ad71a80785937a9732d failed.
+ at
org.apache.flink.client.cli.CliFrontend.triggerSavepoint(CliFrontend. java:828)
+ at
org.apache.flink.client.cli.CliFrontend.lambda$savepopint$8(CliFrontend.java:794)
+ at
org.apache.flink.client.cli.CliFrontend.runClusterAction(CliFrontend.java:1078)
+ at
org.apache.flink.client.cli.CliFrontend.savepoint(CliFrontend.java:779)
+ at
org.apache.flink.client.cli.CliFrontend.parseAndRun(CliFrontend.java:1150)
+ at
org.apache.flink.client.cli.CliFrontend.lambda$mainInternal$9(CliFrontend.java:1226)
+ at
org.apache.flink.runtime.security.contexts.NoOpSecurityContext.runSecured(NoOpSecurityContext.java:28)
+ at
org.apache.flink.client.cli.CliFrontend.mainInternal(CliFrontend.java:1226)
+ at org.apache.flink.client.cli.CliFrontend.main(CliFronhtend.java:1194)
+Caused by: java.util.concurrent.TimeoutException
+ at
java.util.concurrent.CompletableFuture.timedGet(CompletableFuture.java:1784)
+ at
java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1928)
+ at
org.apache.flink.client.cli.CliFrontend.triggerSavepoint(CliFrontend.java:822)
+ ... 8 more
+```
+In this case, we could use "-dsp" option to trigger a detached savepoint, the
client will return immediately as soon as the trigger id returns.
Review Comment:
```suggestion
In this case, we could use "-dsp" option to trigger a detached savepoint,
the client will return the trigger id immediately.
```
##########
docs/content.zh/docs/deployment/cli.md:
##########
@@ -125,6 +125,43 @@ Lastly, you can optionally provide what should be the
[binary format]({{< ref "d
The path to the savepoint can be used later on to [restart the Flink
job](#starting-a-job-from-a-savepoint).
+If the state of the job is quite big, the client will get a timeout exception
since it should wait for the savepoint finished.
+```
+Triggering savepoint for job bec5244e09634ad71a80785937a9732d.
+Waiting for response...
+
+--------------------------------------------------------------
+The program finished with the following exception:
+
+org.apache.flink.util.FlinkException: Triggering a savepoint for the job
bec5244e09634ad71a80785937a9732d failed.
+ at
org.apache.flink.client.cli.CliFrontend.triggerSavepoint(CliFrontend. java:828)
+ at
org.apache.flink.client.cli.CliFrontend.lambda$savepopint$8(CliFrontend.java:794)
+ at
org.apache.flink.client.cli.CliFrontend.runClusterAction(CliFrontend.java:1078)
+ at
org.apache.flink.client.cli.CliFrontend.savepoint(CliFrontend.java:779)
+ at
org.apache.flink.client.cli.CliFrontend.parseAndRun(CliFrontend.java:1150)
+ at
org.apache.flink.client.cli.CliFrontend.lambda$mainInternal$9(CliFrontend.java:1226)
+ at
org.apache.flink.runtime.security.contexts.NoOpSecurityContext.runSecured(NoOpSecurityContext.java:28)
+ at
org.apache.flink.client.cli.CliFrontend.mainInternal(CliFrontend.java:1226)
+ at org.apache.flink.client.cli.CliFrontend.main(CliFronhtend.java:1194)
+Caused by: java.util.concurrent.TimeoutException
+ at
java.util.concurrent.CompletableFuture.timedGet(CompletableFuture.java:1784)
+ at
java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1928)
+ at
org.apache.flink.client.cli.CliFrontend.triggerSavepoint(CliFrontend.java:822)
+ ... 8 more
+```
+In this case, we could use "-dsp" option to trigger a detached savepoint, the
client will return immediately as soon as the trigger id returns.
+```bash
+$ ./bin/flink savepoint \
+ $JOB_ID \
+ /tmp/flink-savepoints
+ -dsp
+```
+```
+Triggering savepoint in detached mode for job bec5244e09634ad71a80785937a9732d.
+Successfully trigger manual savepoint, triggerId:
2505bbd12c5b58fd997d0f193db44b97
+```
+We can get the status of the detached savepoint by rest api
"/jobs/:jobid/savepoints/:triggerid".
Review Comment:
Could we just add a link of REST API here ?
##########
flink-clients/src/main/java/org/apache/flink/client/program/ClusterClient.java:
##########
@@ -188,6 +207,21 @@ CompletableFuture<String> triggerSavepoint(
*/
CompletableFuture<Long> triggerCheckpoint(JobID jobId, CheckpointType
checkpointType);
+ /**
+ * Triggers a detached savepoint for the job identified by the job id. The
savepoint will be
+ * written to the given savepoint directory, or {@link
+ *
org.apache.flink.configuration.CheckpointingOptions#SAVEPOINT_DIRECTORY} if it
is null.
+ * Notice that: detach savepoint will return with a savepoint trigger id
instead of the path
+ * future, that means the client will return very quickly.
+ *
+ * @param jobId job id
+ * @param savepointDirectory directory the savepoint should be written to
+ * @param formatType a binary format of the savepoint
+ * @return The savepoint trigger id
+ */
+ CompletableFuture<String> triggerDetachSavepoint(
Review Comment:
Let's use 'triggerDetachedSavepoint' aligned with
'stopWithDetachedSavepoint'.
You could check other places about 'detach savepoint'.
##########
flink-clients/src/main/java/org/apache/flink/client/program/ClusterClient.java:
##########
@@ -188,6 +207,21 @@ CompletableFuture<String> triggerSavepoint(
*/
CompletableFuture<Long> triggerCheckpoint(JobID jobId, CheckpointType
checkpointType);
+ /**
+ * Triggers a detached savepoint for the job identified by the job id. The
savepoint will be
+ * written to the given savepoint directory, or {@link
+ *
org.apache.flink.configuration.CheckpointingOptions#SAVEPOINT_DIRECTORY} if it
is null.
+ * Notice that: detach savepoint will return with a savepoint trigger id
instead of the path
+ * future, that means the client will return very quickly.
+ *
+ * @param jobId job id
+ * @param savepointDirectory directory the savepoint should be written to
+ * @param formatType a binary format of the savepoint
+ * @return The savepoint trigger id
+ */
+ CompletableFuture<String> triggerDetachSavepoint(
Review Comment:
Let's use 'triggerDetachedSavepoint' aligned with
'stopWithDetachedSavepoint'.
You could check other places about 'detach savepoint'.
##########
flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendSavepointTest.java:
##########
@@ -209,6 +209,34 @@ void testDisposeSavepointSuccess() throws Exception {
}
}
+ // ------------------------------------------------------------------------
+ // detach savepoint
+ // ------------------------------------------------------------------------
+
+ @Test
+ void testTriggerDetachSavepointSuccess() throws Exception {
Review Comment:
Could you also test stop-with-detached-savepoint ?
##########
flink-clients/src/main/java/org/apache/flink/client/program/MiniClusterClient.java:
##########
@@ -115,6 +124,12 @@ public CompletableFuture<Long> triggerCheckpoint(JobID
jobId, CheckpointType che
return miniCluster.triggerCheckpoint(jobId, checkpointType);
}
+ @Override
+ public CompletableFuture<String> triggerDetachSavepoint(
+ JobID jobId, @Nullable String savepointDirectory,
SavepointFormatType formatType) {
+ throw new UnsupportedOperationException();
Review Comment:
Could you implement them simply ?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
