Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/1434#discussion_r46936527
--- 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 {
+ sender() ! TriggerSavepointFailure(jobId, new
IllegalStateException(
+ "Checkpointing disabled. You can enable it via the execution
environment of " +
--- End diff --
This is also breaking the argument list.
---
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 [email protected] or file a JIRA ticket
with INFRA.
---
