If you face any problem while using generic in apache flink :
org.apache.flink.api.common.functions.InvalidTypesException
Exception in thread "main" org.apache.flink.api.common.functions.InvalidTypesException: The return type of function 'main(SourceCategoryMetaRunner.java:63)' could not be determined automatically, due to type erasure. You can give type information hints by using the returns(...) method on the result of the transformation call, or by letting your function implement the 'ResultTypeQueryable' interface.
at org.apache.flink.streaming.api.transformations.StreamTransformation.getOutputType(StreamTransformation.java:262)
at org.apache.flink.streaming.api.datastream.DataStream.addSink(DataStream.java:1063)
at in.dailyhunt.p13n.cis.runner.SourceCategoryMetaRunner.main(SourceCategoryMetaRunner.java:74)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:606)
at com.intellij.rt.execution.application.AppMain.main(AppMain.java:144)
Caused by: org.apache.flink.api.common.functions.InvalidTypesException: Type of TypeVariable 'OUT' in 'class org.apache.flink.api.common.functions.RichMapFunction' could not be determined. This is most likely a type erasure problem. The type extraction currently supports types with generic variables only in cases where all variables in the return type can be deduced from the input type(s).
at org.apache.flink.api.java.typeutils.TypeExtractor.createTypeInfoWithTypeHierarchy(TypeExtractor.java:542)
at org.apache.flink.api.java.typeutils.TypeExtractor.privateCreateTypeInfo(TypeExtractor.java:437)
at org.apache.flink.api.java.typeutils.TypeExtractor.getUnaryOperatorReturnType(TypeExtractor.java:306)
at org.apache.flink.api.java.typeutils.TypeExtractor.getMapReturnTypes(TypeExtractor.java:120)
at org.apache.flink.streaming.api.datastream.DataStream.map(DataStream.java:505)
at in.dailyhunt.p13n.cis.runner.SourceCategoryMetaRunner.main(SourceCategoryMetaRunner.java:63)
... 5 more
Process finished with exit code 1
FOR Example :
You have Map Function
public class WorkFlowProcessor<T extends Enrichable> extends RichMapFunction<String, T> {
private WorkflowConfig workflowConfig;
private transient Workflow workflow;
private Map<String, String> config;
public WorkFlowProcessor(WorkflowConfig workflowConfig, Map<String, String> config) {
this.workflowConfig = workflowConfig;
this.config = config;
}
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
this.workflow = new Workflow(workflowConfig, config);
}
@Override
public T map(String input) throws Exception {
return (T) this.workflow.execute(input.toLowerCase());
}
and from Main class your are passing '
WorkFlowProcessor<MetaEntity> workFlowProcessor = new WorkFlowProcessor<>(workflowConfig, config.toMap());
DataStream<MetaEntity> enrichables = data.map(workFlowProcessor);
where WorkFlowProcessor is map function .
When you will run it Type information is erased and exception will be thrown.
To Solve this you can add return type like returns(MetaEntity.class); there are called as Type Hints in the Java API
DataStream<MetaEntity> enrichables = data.map(workFlowProcessor).returns(MetaEntity.class);
Helpful Links :