apache flink with Kafka: InvalidTypesException

Question

I have following code:

Properties properties = new Properties();
properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, MyCustomClassDeserializer.class.getName());

FlinkKafkaConsumer<MyCustomClass> kafkaConsumer = new FlinkKafkaConsumer(
                    "test-kafka-topic",
                    new SimpleStringSchema(),
                    properties);

final StreamExecutionEnvironment streamEnv = StreamExecutionEnvironment.getExecutionEnvironment();
DataStream<MyCustomClass> kafkaInputStream = streamEnv.addSource(kafkaConsumer);

DataStream<String> stringStream = kafkaInputStream
                    .map(new MapFunction<MyCustomClass,String>() {
                        @Override
                        public String map(MyCustomClass message) {
                            logger.info("--- Received message : " + message.toString());
                            return message.toString();
                        }
                    });

streamEnv.execute("Published messages");

MyCustomClassDeserializer is implemented to convert byte array to java object.

When I run this program locally, I get error:

Caused by: org.apache.flink.api.common.functions.InvalidTypesException: Input mismatch: Basic type expected.

And I get this for code line:

.map(new MapFunction<MyCustomClass,String>() {

Not sure why I get this?

Dominik Wosiński · Accepted Answer · 2020-05-04 10:07:15Z

1

So, You have a deserializer that returns POJO, yet You are telling Flink that it should deserialize record from byte[] to String by using SimpleStringSchema. See the problem now? :)

I don't think You should use the custom Kafka deserializers in FlinkKafkaConsumer in general. What You should aim for instead is to instead create a custom class that extends DeserializationSchema from Flink. It should be much better in terms of type safety and testability.

answered May 4, 2020 at 10:07

Dominik Wosiński

3,8941 gold badge11 silver badges22 bronze badges

Sign up to request clarification or add additional context in comments.

1 Comment

Mandroid Over a year ago

Thanks, got it. Actually its a legacy code, with too many such issues.

Collectives™ on Stack Overflow

apache flink with Kafka: InvalidTypesException

1 Answer 1

1 Comment

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

1 Comment

Your Answer

Sign up or log in

Post as a guest

Related