State & Generics

Posted by Laurent Exsteens on
URL: http://deprecated-apache-flink-user-mailing-list-archive.369.s1.nabble.com/State-Generics-tp33956.html

Hello,

Using Flink 1.8.1, I'm getting the following error:

     The TypeHint is using a generic variable.This is not supported, generic types must be fully specified for the TypeHint.

when trying to create a ListStateDescriptor with a generic type (full sample code in attachment):

public class AND<LEFT, RIGHT> extends RichCoFlatMapFunction<LEFT, RIGHT, Tuple2<LEFT, RIGHT>> {

    private transient ListState<LEFT> leftState;
    private transient ListState<RIGHT> rightState;

    @Override
    public void open(Configuration config) {
        ListStateDescriptor<LEFT> left_descriptor =
            new ListStateDescriptor<>(
                "and_left",
                TypeInformation.of(new TypeHint<LEFT>() {
                }));
        leftState = getRuntimeContext().getListState(left_descriptor);

This gives me the following stack trace:

Exception in thread "main" org.apache.flink.runtime.client.JobExecutionException: Job execution failed.
at org.apache.flink.runtime.jobmaster.JobResult.toJobExecutionResult(JobResult.java:146)
at org.apache.flink.runtime.minicluster.MiniCluster.executeJobBlocking(MiniCluster.java:638)
at org.apache.flink.streaming.api.environment.LocalStreamEnvironment.execute(LocalStreamEnvironment.java:123)
at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.java:1509)
at eu.euranova.leadcep.Main.main(Main.java:61)
Caused by: org.apache.flink.util.FlinkRuntimeException: The TypeHint is using a generic variable.This is not supported, generic types must be fully specified for the TypeHint.
at org.apache.flink.api.common.typeinfo.TypeHint.<init>(TypeHint.java:54)
at eu.euranova.leadcep.AND$1.<init>(AND.java:22)
at eu.euranova.leadcep.AND.open(AND.java:19)
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.api.operators.co.CoStreamFlatMap.open(CoStreamFlatMap.java:46)
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)

Googling the error didn't give a working solution.

Is there a way to work with state using Generic types? if yes, how?

Thanks in advance for your help!

Best Regards,

Laurent.

--

♻ Be green, keep it on the screen