wang created FLINK-19866:
----------------------------
Summary: FunctionsStateBootstrapOperator.createStateAccessor fails
due to uninitialized runtimeContext
Key: FLINK-19866
URL: https://issues.apache.org/jira/browse/FLINK-19866
Project: Flink
Issue Type: Bug
Components: Stateful Functions
Affects Versions: 1.11.2, statefun-2.2.0
Reporter: wang
It has bugs similar to
[FLINK-19330|https://issues.apache.org/jira/browse/FLINK-19330]
In Flink 1.11.2, statefun-flink-state-processor 2.2.0, the
AbstractStreamOperator's runtimeContext is not fully initialized when executing
AbstractStreamOperator#intializeState()
in particular KeyedStateStore is set after intializeState was finished.
See:
[https://github.com/apache/flink/blob/master/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java#L258,L259]
This behaviour was changed from Flink 1.10->Flink 1.11.
StateFun's FunctionsStateBootstrapOperator performs its initialization logic at
initalizeState, and it requires an already initialized runtimeContext to create
stateAccessor.
This situation causes the following failure:
{code:java}
Caused by: java.lang.NullPointerException: Keyed state can only be used on a
'keyed stream', i.e., after a 'keyBy()' operation.Caused by:
java.lang.NullPointerException: Keyed state can only be used on a 'keyed
stream', i.e., after a 'keyBy()' operation. at
org.apache.flink.util.Preconditions.checkNotNull(Preconditions.java:75) at
org.apache.flink.streaming.api.operators.StreamingRuntimeContext.checkPreconditionsAndGetKeyedStateStore(StreamingRuntimeContext.java:223)
at
org.apache.flink.streaming.api.operators.StreamingRuntimeContext.getState(StreamingRuntimeContext.java:188)
at
org.apache.flink.statefun.flink.core.state.FlinkState.createFlinkStateAccessor(FlinkState.java:69)
at
org.apache.flink.statefun.flink.core.state.FlinkStateBinder.bindValue(FlinkStateBinder.java:48)
at org.apache.flink.statefun.sdk.state.StateBinder.bind(StateBinder.java:30)
at
org.apache.flink.statefun.flink.core.state.PersistedStates.findReflectivelyAndBind(PersistedStates.java:46)
at
org.apache.flink.statefun.flink.state.processor.operator.StateBootstrapFunctionRegistry.bindState(StateBootstrapFunctionRegistry.java:120)
at
org.apache.flink.statefun.flink.state.processor.operator.StateBootstrapFunctionRegistry.initialize(StateBootstrapFunctionRegistry.java:103)
at
org.apache.flink.statefun.flink.state.processor.operator.StateBootstrapper.<init>(StateBootstrapper.java:39)
at
org.apache.flink.statefun.flink.state.processor.operator.FunctionsStateBootstrapOperator.initializeState(FunctionsStateBootstrapOperator.java:67)
at
org.apache.flink.streaming.api.operators.StreamOperatorStateHandler.initializeOperatorState(StreamOperatorStateHandler.java:106)
at
org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:258)
at
org.apache.flink.state.api.output.BoundedStreamTask.init(BoundedStreamTask.java:85)
at
org.apache.flink.streaming.runtime.tasks.StreamTask.beforeInvoke(StreamTask.java:457)
at
org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:522)
at
org.apache.flink.state.api.output.BoundedOneInputStreamTaskRunner.mapPartition(BoundedOneInputStreamTaskRunner.java:76)
at
org.apache.flink.runtime.operators.MapPartitionDriver.run(MapPartitionDriver.java:103)
at org.apache.flink.runtime.operators.BatchTask.run(BatchTask.java:504) at
org.apache.flink.runtime.operators.BatchTask.invoke(BatchTask.java:369) at
org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:721) at
org.apache.flink.runtime.taskmanager.Task.run(Task.java:546) at
java.lang.Thread.run(Thread.java:748){code}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
