Github user uce commented on a diff in the pull request:
https://github.com/apache/flink/pull/1633#discussion_r53000913
--- Diff:
flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
---
@@ -1073,57 +1073,73 @@ class JobManager(
// execute the recovery/writing the jobGraph into the
SubmittedJobGraphStore asynchronously
// because it is a blocking operation
future {
- try {
- if (isRecovery) {
- executionGraph.restoreLatestCheckpointedState()
- }
- else {
- val snapshotSettings = jobGraph.getSnapshotSettings
- if (snapshotSettings != null) {
- val savepointPath = snapshotSettings.getSavepointPath()
+ val restoreStateSuccess =
+ try {
+ if (isRecovery) {
+ executionGraph.restoreLatestCheckpointedState()
--- End diff --
Regarding the `JobSubmitSuccess`: we had it as a follow up to have more
fine-grained integration with the the client and left it as a duplicate submit
message for the time being (instead of something like `JobRecovered`).
The other behaviour is back to the previous state now. I hear you that it
makes sense to integrate the state restore behaviour with the execution graph
restart.
---
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.
---
