Hi,
When I run the code implementing a generic FlatMapFunction, Flink
complained about InvalidTypesException:
public class GenericFlatMapper<T> implements
FlatMapFunction<SortedMap<T, Long>, Tuple2<T, Long>> {
@Override
public void flatMap(SortedMap<T, Long> m, Collector<Tuple2<T,
Long>> out) throws Exception {
for (Map.Entry<T, Long> entry : m.entrySet()) {
out.collect(Tuple2.of(entry.getKey(), entry.getValue()));
}
}
}
*Exception in thread "main"
org.apache.flink.api.common.functions.InvalidTypesException: The return
type of function 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.*
*...*
*Caused by: org.apache.flink.api.common.functions.InvalidTypesException:
Type of TypeVariable 'T' in 'class GenericFlatMapper' 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).*
This puzzles me as Flink should be able to infer the type from arguments. I
know returns(...) or other workarounds to give type hint, but they are kind
of verbose. Any suggestions?
