Apache Flink Kafka集成

Question

我正在尝试将Apache Kafka 2.11-0.10.0.0与Apache Flink 1.1.2集成。我正在使用scala-shell对其进行测试，并得到以下错误。

找不到类org.apache.flink.streaming.api.checkpoint.CheckpointNotifier

我已经将org.apache.flink.streaming jar添加到类路径中，但这没有帮助。我一直导入到org.apache.flink.streaming.api.checkpoint._。这仍然没有帮助。下面是我在shell中运行的代码

 import org.apache.flink.streaming.connectors.kafka._
 import org.apache.flink.streaming.util.serialization.SimpleStringSchema
 import org.apache.flink._
 import java.util._
 val properties = new Properties()
 properties.setProperty("bootstrap.servers", "localhost:9092")
 properties.setProperty("zookeeper.connect", "localhost:2181")
 properties.setProperty("group.id", "test")
 val myFetcher = FlinkKafkaConsumer.FetcherType.NEW_HIGH_LEVEL
 val myHandler = FlinkKafkaConsumer.OffsetStore.FLINK_ZOOKEEPER
 senv.addSource(new FlinkKafkaConsumer[String]("topic", new SimpleStringSchema(), properties, myHandler, myFetcher)).print

我还尝试使用FlinkKafkaConsumer081和FlinkKafkaConsumer082方法代替FlinkKafkaConsumer方法，但仍然收到相同的错误。

我确实注意到flink-connector-kafka jar自2月16日以来就没有更新过。这是不是用错了jar？我确实在maven central Flink Connector中找到了Kafka Base 2_11。我应该使用那个jar吗？

请帮帮我！

Answer 1

您存在版本冲突。这个类在Flink 1.0中被删除了，我相信。所以你可能有一个旧版本的jar。检查以确保所有内容都已更新到Flink 1.1.2