Flink KafkaSource编译报错：valueOnly方法参数不匹配求助

我想要搭建一个通过Kafka向Flink传输数据流的程序，编写了如下代码：

DataStream<Integer> input =
        env.fromSource(
                KafkaSource.<Integer>builder()
                        .setBootstrapServers(
                                parameterTool
                                        .getProperties()
                                        .getProperty(
                                                ProducerConfig.BOOTSTRAP_SERVERS_CONFIG))
                        .setBounded(OffsetsInitializer.latest())
                        .setDeserializer(
                                KafkaRecordDeserializationSchema.valueOnly(
                                        IntegerDeserializer.class))
                        .setTopics(parameterTool.getRequired("input-topic"))
                        .build(),
                WatermarkStrategy.noWatermarks(),
                "kafka-source");

编译时触发错误：

Exception in thread "main" java.lang.Error: Unresolved compilation problem: 
    The method valueOnly(DeserializationSchema<V>) in the type KafkaRecordDeserializationSchema is not applicable for the arguments (Class<NumberDeserializers.IntegerDeserializer>)

问题出在valueOnly方法调用上，不清楚具体原因和解决办法。

核心原因 #

KafkaRecordDeserializationSchema.valueOnly()有两个重载版本，但你传入的IntegerDeserializer.class是Class对象，和任何一个重载的参数类型都不匹配：

• 第一个重载要求传入Flink的DeserializationSchema<V>实例
• 第二个重载要求传入Kafka的Deserializer<V>实例

另外要注意：你用的IntegerDeserializer是Kafka自带的反序列化器（属于org.apache.kafka.common.serialization包），并非Flink的DeserializationSchema接口实现类。

两种解决办法 #

办法一：传入Kafka反序列化器实例

直接创建IntegerDeserializer的对象传入，满足第二个重载的参数要求：

.setDeserializer(
        KafkaRecordDeserializationSchema.valueOnly(
                new IntegerDeserializer()))

办法二：使用Flink的反序列化Schema

如果想基于Flink的类型体系实现反序列化，可以用TypeInformationDeserializationSchema：

import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.serialization.TypeInformationDeserializationSchema;

// 替换原有setDeserializer代码
.setDeserializer(
        KafkaRecordDeserializationSchema.valueOnly(
                new TypeInformationDeserializationSchema<>(TypeInformation.of(Integer.class))))

额外提醒 #

确保导入的是正确的Kafka反序列化器类：

import org.apache.kafka.common.serialization.IntegerDeserializer;

内容的提问来源于stack exchange，提问作者Mike