Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/1434#discussion_r46936461
--- Diff:
flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
---
@@ -495,6 +513,76 @@ class JobManager(
case checkpointMessage : AbstractCheckpointMessage =>
handleCheckpointMessage(checkpointMessage)
+ case TriggerSavepoint(jobId) =>
+ currentJobs.get(jobId) match {
+ case Some((graph, _)) =>
+ val savepointCoordinator = graph.getSavepointCoordinator()
+
+ if (savepointCoordinator != null) {
+ // Immutable copy for the future
+ val senderRef = sender()
+
+ future {
+ try {
+ // Do this async, because checkpoint coordinator
operations can
+ // contain blocking calls to the state backend or
ZooKeeper.
+ val savepointFuture =
savepointCoordinator.triggerSavepoint(
+ System.currentTimeMillis())
+
+ savepointFuture.onComplete {
+ // Success, respond with the savepoint path
+ case scala.util.Success(savepointPath) =>
+ senderRef ! TriggerSavepointSuccess(jobId,
savepointPath)
+
+ // Failure, respond with the cause
+ case scala.util.Failure(t) =>
+ senderRef ! TriggerSavepointFailure(jobId,
--- End diff --
If you break argument lists, then every argument is written in its own line.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
