[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.
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 ec33811..7df4194 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 @@
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;
@@ -61,11 +60,12 @@
}
@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
