[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 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 @@
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 @@
}
@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
