Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/1434#discussion_r46850418
--- 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,
+ new Exception("Failed to complete savepoint", t))
+ }(context.dispatcher)
+ }
+ catch {
+ case e: Exception =>
+ senderRef ! TriggerSavepointFailure(jobId, new Exception(
+ "Failed to trigger savepoint", e))
+ }
+ }(context.dispatcher)
+ }
+ else {
--- End diff --
The same for `else`.
---
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.
---
