[FLINK-19330][core] 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 #159.
diff --git a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/functions/FunctionGroupOperator.java b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/functions/FunctionGroupOperator.java
index cbb9131..9cc0818 100644
--- a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/functions/FunctionGroupOperator.java
+++ b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/functions/FunctionGroupOperator.java
@@ -27,7 +27,6 @@
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.common.typeutils.base.LongSerializer;
import org.apache.flink.runtime.state.KeyedStateBackend;
-import org.apache.flink.runtime.state.StateInitializationContext;
import org.apache.flink.runtime.state.StateSnapshotContext;
import org.apache.flink.runtime.state.internal.InternalListState;
import org.apache.flink.statefun.flink.core.StatefulFunctionsConfig;
@@ -93,9 +92,8 @@
}
@Override
- public void initializeState(StateInitializationContext context) throws Exception {
- super.initializeState(context);
-
+ public void open() throws Exception {
+ super.open();
final StatefulFunctionsUniverse statefulFunctionsUniverse =
statefulFunctionsUniverse(configuration);
@@ -108,7 +106,7 @@
new ListStateDescriptor<>(
FlinkStateDelayedMessagesBuffer.BUFFER_STATE_NAME, envelopeSerializer.duplicate());
final MapState<Long, Message> asyncOperationState =
- context.getKeyedStateStore().getMapState(asyncOperationStateDescriptor);
+ getRuntimeContext().getMapState(asyncOperationStateDescriptor);
Objects.requireNonNull(mailboxExecutor, "MailboxExecutor is unexpectedly NULL");
