You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tz...@apache.org on 2020/11/04 06:14:42 UTC
[flink-statefun] branch release-2.2 updated: [FLINK-19866] Move
intialization logic to open() instead initializeState
This is an automated email from the ASF dual-hosted git repository.
tzulitai pushed a commit to branch release-2.2
in repository https://gitbox.apache.org/repos/asf/flink-statefun.git
The following commit(s) were added to refs/heads/release-2.2 by this push:
new 653393f [FLINK-19866] Move intialization logic to open() instead initializeState
653393f is described below
commit 653393fcfb9242d0d9169e4b2133de8bf5a08399
Author: wangchao <wa...@apexsoft.com.cn>
AuthorDate: Fri Oct 30 11:01:25 2020 +0800
[FLINK-19866] Move intialization logic to open() instead initializeState
This commit moves the initialization logic of the operator to open()
instead of initalizeState(). The reason is that starting from Flink 1.10,
at intializeState the runtimeContext is not yet properly initialized.
This closes #170.
---
.../processor/operator/FunctionsStateBootstrapOperator.java | 12 ++++++------
1 file changed, 6 insertions(+), 6 deletions(-)
diff --git a/statefun-flink/statefun-flink-state-processor/src/main/java/org/apache/flink/statefun/flink/state/processor/operator/FunctionsStateBootstrapOperator.java b/statefun-flink/statefun-flink-state-processor/src/main/java/org/apache/flink/statefun/flink/state/processor/operator/FunctionsStateBootstrapOperator.java
index 09e6b59..c2e14d6 100644
--- a/statefun-flink/statefun-flink-state-processor/src/main/java/org/apache/flink/statefun/flink/state/processor/operator/FunctionsStateBootstrapOperator.java
+++ b/statefun-flink/statefun-flink-state-processor/src/main/java/org/apache/flink/statefun/flink/state/processor/operator/FunctionsStateBootstrapOperator.java
@@ -20,7 +20,6 @@ package org.apache.flink.statefun.flink.state.processor.operator;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.core.fs.Path;
import org.apache.flink.runtime.state.KeyedStateBackend;
-import org.apache.flink.runtime.state.StateInitializationContext;
import org.apache.flink.state.api.output.SnapshotUtils;
import org.apache.flink.state.api.output.TaggedOperatorSubtaskState;
import org.apache.flink.statefun.flink.core.functions.FunctionGroupOperator;
@@ -60,11 +59,12 @@ public final class FunctionsStateBootstrapOperator
}
@Override
- public void initializeState(StateInitializationContext context) throws Exception {
- super.initializeState(context);
-
- final State stateAccessor = createStateAccessor(getRuntimeContext(), getKeyedStateBackend());
- this.stateBootstrapper = new StateBootstrapper(stateBootstrapFunctionRegistry, stateAccessor);
+ public void open() throws Exception {
+ super.open();
+ if (this.stateBootstrapper == null) {
+ final State stateAccessor = createStateAccessor(getRuntimeContext(), getKeyedStateBackend());
+ this.stateBootstrapper = new StateBootstrapper(stateBootstrapFunctionRegistry, stateAccessor);
+ }
}
@Override