我对斯卡拉和Flink还不熟悉。我使用下面的代码来使用来自kafka流的json消息。

object Job {
  def main(args: Array[String]) {
    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "localhost:9092")
    // only required for Kafka 0.8
    properties.setProperty("zookeeper.connect", "localhost:2181")
    val topic: String = "test"

    val consumer = new FlinkKafkaConsumer[ObjectNode](topic, new JSONKeyValueDeserializationSchema(false), properties)
    // set up the execution environment
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val stream = env.addSource(consumer).name("opendental")
      .enableCheckpointing(5000) // checkpoint every 5000 msecs
      .print()

    // execute program
    env.execute("Flink Scala API Skeleton")
  }
}

我在使用命令 mvn clean package 将scala代码编译为 .jar 但是由于重载方法，生成失败。这就是错误

[ERROR]   (x$1: java.util.regex.Pattern,x$2: org.apache.flink.streaming.util.serialization.KeyedDeserializationSchema[com.fasterxml.jackson.databind.node.ObjectNode],x$3: java.util.Properties)org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer[com.fasterxml.jackson.databind.node.ObjectNode] <and>
[ERROR]   (x$1: java.util.regex.Pattern,x$2: (some other)org.apache.flink.api.common.serialization.DeserializationSchema(in <none>)[com.fasterxml.jackson.databind.node.ObjectNode],x$3: java.util.Properties)org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer[com.fasterxml.jackson.databind.node.ObjectNode] <and>
[ERROR]   (x$1: java.util.List[String],x$2: org.apache.flink.streaming.util.serialization.KeyedDeserializationSchema[com.fasterxml.jackson.databind.node.ObjectNode],x$3: java.util.Properties)org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer[com.fasterxml.jackson.databind.node.ObjectNode] <and>
[ERROR]   (x$1: java.util.List[String],x$2: org.apache.flink.api.common.serialization.DeserializationSchema(in <none>)[com.fasterxml.jackson.databind.node.ObjectNode],x$3: java.util.Properties)org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer[com.fasterxml.jackson.databind.node.ObjectNode] <and>
[ERROR]   (x$1: String,x$2: org.apache.flink.streaming.util.serialization.KeyedDeserializationSchema[com.fasterxml.jackson.databind.node.ObjectNode],x$3: java.util.Properties)org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer[com.fasterxml.jackson.databind.node.ObjectNode] <and>
[ERROR]   (x$1: String,x$2: org.apache.flink.api.common.serialization.DeserializationSchema[com.fasterxml.jackson.databind.node.ObjectNode],x$3: java.util.Properties)org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer[com.fasterxml.jackson.databind.node.ObjectNode]
[ERROR]  cannot be applied to (String, org.apache.flink.streaming.util.serialization.JSONKeyValueDeserializationSchema, java.util.Properties)
[ERROR]     val consumer = new FlinkKafkaConsumer[ObjectNode](topic, new JSONKeyValueDeserializationSchema(false), properties)
[ERROR]

我尝试过几种方法来解决重载方法，但现在遇到了麻烦。