ClassNotFoundException: org.apache.flink.streaming.api.checkpoint.CheckpointNotifier при использовании темы kafka

Я использую новейшие банки Flink-1.1.2-Hadoop-27 и flink-connector-kafka-0.10.2-hadoop1.

Потребитель Flink выглядит следующим образом:

StreamExecutionEnvironment env=StreamExecutionEnvironment.getExecutionEnvironment();
        if (properties == null) {
            properties = new Properties();
            InputStream props = Resources.getResource(KAFKA_CONFIGURATION_FILE).openStream();
            properties.load(props);

            DataStream<String> stream = env.addSource(new FlinkKafkaConsumer082<>(KAFKA_SIP_TOPIC, new SimpleStringSchema() , properties));

Ниже приводится исключение, которое я получаю после выполнения:

Exception in thread "main" java.lang.NoClassDefFoundError: org/apache/flink/streaming/api/checkpoint/CheckpointNotifier
at java.lang.ClassLoader.defineClass1(Native Method)
at java.lang.ClassLoader.defineClass(Unknown Source)
at java.security.SecureClassLoader.defineClass(Unknown Source)
at java.net.URLClassLoader.defineClass(Unknown Source)
at java.net.URLClassLoader.access$100(Unknown Source)
at java.net.URLClassLoader$1.run(Unknown Source)
at java.net.URLClassLoader$1.run(Unknown Source)
at java.security.AccessController.doPrivileged(Native Method)
at java.net.URLClassLoader.findClass(Unknown Source)
at java.lang.ClassLoader.loadClass(Unknown Source)
at sun.misc.Launcher$AppClassLoader.loadClass(Unknown Source)
at java.lang.ClassLoader.loadClass(Unknown Source)
at java.lang.ClassLoader.defineClass1(Native Method)
at java.lang.ClassLoader.defineClass(Unknown Source)
at java.security.SecureClassLoader.defineClass(Unknown Source)
at java.net.URLClassLoader.defineClass(Unknown Source)
at java.net.URLClassLoader.access$100(Unknown Source)
at java.net.URLClassLoader$1.run(Unknown Source)
at java.net.URLClassLoader$1.run(Unknown Source)
at java.security.AccessController.doPrivileged(Native Method)
at java.net.URLClassLoader.findClass(Unknown Source)
at java.lang.ClassLoader.loadClass(Unknown Source)
at sun.misc.Launcher$AppClassLoader.loadClass(Unknown Source)
at java.lang.ClassLoader.loadClass(Unknown Source)
at com.bt.oss.voice.main.FlnkConsumer.main(FlnkConsumer.java:50)Caused by: java.lang.ClassNotFoundException: org.apache.flink.streaming.api.checkpoint.CheckpointNotifier
at java.net.URLClassLoader.findClass(Unknown Source)
at java.lang.ClassLoader.loadClass(Unknown Source)
at sun.misc.Launcher$AppClassLoader.loadClass(Unknown Source)
at java.lang.ClassLoader.loadClass(Unknown Source)
... 25 more

apache-kafka apache-flink flink-streaming
person mrinal    schedule 14.10.2016    source источник

Ответы (2)


arrow_upward
8
arrow_downward

Вы смешиваете версии. Потребитель Kafka для Flink 0.10.2 не будет работать с Flink 1.1.2.

Вы должны использовать коннектор Kafka, поставляемый с Flink 1.1.2, и включить следующую зависимость Maven:

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-connector-kafka-0.8_2.10</artifactId>
  <version>1.1.2</version>
</dependency>

Ознакомьтесь с документацией для подробностей.

person Fabian Hueske    schedule 14.10.2016

arrow_upward
1
arrow_downward

В случае использования sbt следующее обновление в зависимостях flink scala решило мою проблему.

"org.apache.flink" %% "flink-scala" % flinkVersion % "provided",
"org.apache.flink" %% "flink-streaming-scala" % flinkVersion % "provided",

to:

"org.apache.flink" %% "flink-scala" % flinkVersion,
"org.apache.flink" %% "flink-streaming-scala" % flinkVersion,
person Evhz    schedule 17.05.2019