git.net

[Date Prev][Date Next][Thread Prev][Thread Next][Date Index][Thread Index]

Re: Initializing mapstate hangs


2018-10-22 13:46:31,944 INFO  org.apache.flink.runtime.taskmanager.Task                     - Window(SlidingProcessingTimeWindows(180000, 180000), TimeTrigger, MetricWindowFunction) -> Map -> Sink: Unnamed (1/1) (5677190a0d292df3ad8f3521519cd980) switched from RUNNING to FAILED.

java.lang.NullPointerException: The state properties must not be null

at org.apache.flink.util.Preconditions.checkNotNull(Preconditions.java:75)

at org.apache.flink.streaming.api.operators.StreamingRuntimeContext.checkPreconditionsAndGetKeyedStateStore(StreamingRuntimeContext.java:174)

at org.apache.flink.streaming.api.operators.StreamingRuntimeContext.getMapState(StreamingRuntimeContext.java:168)

at com.sap.hybris.conversion.flink.processors.chain.MetricWindowFunction.open(MetricWindowFunction.java:62)

at org.apache.flink.api.common.functions.util.FunctionUtils.openFunction(FunctionUtils.java:36)

at org.apache.flink.api.java.operators.translation.WrappingFunction.open(WrappingFunction.java:45)

at org.apache.flink.api.common.functions.util.FunctionUtils.openFunction(FunctionUtils.java:36)

at org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.open(AbstractUdfStreamOperator.java:102)

at org.apache.flink.streaming.runtime.operators.windowing.WindowOperator.open(WindowOperator.java:219)

at org.apache.flink.streaming.runtime.tasks.StreamTask.openAllOperators(StreamTask.java:424)

at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:290)

at org.apache.flink.runtime.taskmanager.Task.run(Task.java:711)

at java.lang.Thread.run(Thread.java:745)





On Sat, 20 Oct 2018 at 11:29, vino yang <yanghua1127@xxxxxxxxx> wrote:
Hi Ahmad,

Can you try to dump thread info from the Task Manager's JVM instance?

Thanks, vino.

Ahmad Hassan <ahmad.hassan@xxxxxxxxx> 于2018年10月20日周六 下午4:24写道:
Flink 1.6.0. Valuestate initialises successful but mapstate hangs 

Regards 

On 20 Oct 2018, at 02:55, vino yang <yanghua1127@xxxxxxxxx> wrote:

Hi Ahmad,

Which version of Flink do you use?

Thanks, vino.

Ahmad Hassan <ahmad.hassan@xxxxxxxxx> 于2018年10月19日周五 下午11:32写道:
Hi,

Initializing mapstate hangs in window function. However if i use valuestate then it is initialized succcessfully. I am using rocksdb to store the state.

public class MyWindowFunction extends RichWindowFunction<Event, Payload, Tuple, TimeWindow>
{
private transient MapStateDescriptor<String, String> productsDescriptor = new MapStateDescriptor<>(
"mapState", String.class, String.class);

@Override
public void apply(Tuple key, TimeWindow window, final Iterable<Event> input,
final Collector<Payload> out)
{
// do something
}

@Override
public void open(Configuration parameters) throws Exception
{
System.out.println("## open init window state ");
MapState<String, String> state = this.getRuntimeContext().getMapState(productsDescriptor); <<< program hangs here
System.out.println("## open window state " + state);
}
}

Thanks for the help.