You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by sj...@apache.org on 2021/02/11 19:46:43 UTC
[flink] 02/02: [FLINK-21315][state-processor-api]set an operator
name when collecting existing operator states.
This is an automated email from the ASF dual-hosted git repository.
sjwiesman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit ae5bea39491860ccdff3316877ef28e64f466f64
Author: Jun Qin <11...@users.noreply.github.com>
AuthorDate: Tue Feb 9 11:38:01 2021 +0100
[FLINK-21315][state-processor-api]set an operator name when collecting existing operator states.
This closes #14907
---
.../src/main/java/org/apache/flink/state/api/WritableSavepoint.java | 5 ++++-
1 file changed, 4 insertions(+), 1 deletion(-)
diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/WritableSavepoint.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/WritableSavepoint.java
index f428a27..324e4c1 100644
--- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/WritableSavepoint.java
+++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/WritableSavepoint.java
@@ -105,7 +105,10 @@ public abstract class WritableSavepoint<F extends WritableSavepoint> {
finalOperatorStates = newOperatorStates;
} else {
DataSet<OperatorState> existingOperatorStates =
- newOperatorStates.getExecutionEnvironment().fromCollection(existingOperators);
+ newOperatorStates
+ .getExecutionEnvironment()
+ .fromCollection(existingOperators)
+ .name("existingOperatorStates");
existingOperatorStates
.flatMap(new StatePathExtractor())