尝试运行简单的 Kafka Stream App 时收到异常答案

【问题标题】：Receiving exception when trying to run simple Kafka Stream App尝试运行简单的 Kafka Stream App 时收到异常
【发布时间】：2019-05-13 12:44:14
【问题描述】：

我一直在尝试使用 Kafka 运行一个简单的 wordcount 应用程序，但是每当我运行它时，我都会收到以下错误：

Exception in thread "main" java.lang.NoClassDefFoundError: org/apache/kafka/common/utils/LogContext
    at org.apache.kafka.streams.KafkaStreams.<init>(KafkaStreams.java:630)
    at org.apache.kafka.streams.KafkaStreams.<init>(KafkaStreams.java:610)
    at org.apache.kafka.streams.KafkaStreams.<init>(KafkaStreams.java:557)
    at StreamsApp.main(StreamsApp.java:49)
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.base/java.lang.reflect.Method.invoke(Method.java:566)
    at com.intellij.rt.execution.application.AppMainV2.main(AppMainV2.java:131)
Caused by: java.lang.ClassNotFoundException: org.apache.kafka.common.utils.LogContext
    at java.base/jdk.internal.loader.BuiltinClassLoader.loadClass(BuiltinClassLoader.java:583)
    at java.base/jdk.internal.loader.ClassLoaders$AppClassLoader.loadClass(ClassLoaders.java:178)
    at java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:521)

我不知道为什么我不断收到此错误... 下面列出了 main 方法的代码。（第 49 行）KafkaStreams 流 = new KafkaStreams(topology, props);

public static void main(final String[] args) throws Exception {

    Properties props = new Properties();
    props.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount-application");
    props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
    props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
    // props.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory().getAbsolutePath());

    StreamsBuilder builder = new StreamsBuilder();
    KStream<String, String> textLines = builder.stream("inputTopic");
    final Pattern pattern = Pattern.compile("\\W+", Pattern.UNICODE_CHARACTER_CLASS);

    KTable<String, Long> wordCounts = textLines
            .flatMapValues(value -> Arrays.asList(pattern.split(value.toLowerCase())))
            .groupBy((key, word) -> word)
            .count();

    wordCounts
            .foreach((w, c) -> System.out.println("word: " + w + " -> " + c));


    String outputTopic = "outputTopic";
    Serde<String> stringSerde = Serdes.String();
    Serde<Long> longSerde = Serdes.Long();
    wordCounts.to(stringSerde, longSerde, outputTopic);

    Topology topology = builder.build();

    KafkaStreams streams = new KafkaStreams(topology, props);
    streams.start();

    Thread.sleep(30000);
    streams.close();

}

}

【问题讨论】：

似乎这是由于您的依赖冲突。请发布您使用的kafka-streams 和kafka-client 的哪个版本？
卡夫卡流 1.1.1
kafka-client 0.11.0.1
这是问题所在。 kafka-streams 版本为 1.1.1，预计 kafka-client 版本高于 1.0.0。所以你需要更新 kafka-client 到 1.1.1

标签： java apache-kafka apache-kafka-streams

【解决方案1】：

您在依赖关系 org.apache.kafka:kafka-streams 和 org.apache.kafka:kafka-clients 之间存在冲突。根据您的例外情况，您使用的kafka-clients 版本小于1.0.0，但kafka-streams 版本等于或高于1.0.0。

确保kafka-clients 版本至少为1.0.0（因此您需要升级您的kafka-clients 版本），否则您需要降级kafka-streams 版本。

【讨论】：

【解决方案2】：

所需的 jars 应该在类路径中。希望以下内容对您有所帮助。

C:\PROJECT\ENVIRONMENT\KAFKA_2_11\bin\windows\kafka-run-class.bat -cp  C:\PROJECT\KafkaDemos\target\classes;C:\PROJECT\ENVIRONMENT\KAFKA_2_11\libs\kafka_2.11-2.4.0.jar;C:\PROJECT\ENVIRONMENT\KAFKA_2_11\libs\kafka-clients-2.4.0.jar;C:\PROJECT\E
NVIRONMENT\KAFKA_2_11\libs\kafka-streams-2.4.0.jar;C:\PROJECT\ENVIRONMENT\KAFKA_2_11\libs\kafka-clients-2.4.0.jar;C:\PROJECT\ENVIRONMENT\KAFKA_2_11\libs\slf4j-api-1.7.28.jar;C:\PROJECT\ENVIRONMENT\ExtraJars\apache-logging-log4j.jar;C:\PROJECT\ENVIRONMENT\ExtraJars\
rocksdbjni-6.5.3.jar; demo.wordcount.WordCount

【讨论】：