java.lang.ClassNotFoundException: org.apache.kafka.common.metrics.MetricsContext

Question

现在我正在学习卡夫卡。根据视频课程，我正在尝试使用 Kafka Streams 构建 Twitter 推文过滤器：

public class StreamsFilterTweet {

    public static void main(String[] args) {

        // create properties
        Properties properties = new Properties();
        properties.setProperty(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
        properties.setProperty(StreamsConfig.APPLICATION_ID_CONFIG, "demo-kafka-streams");
        properties.setProperty(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.StringSerde.class.getName());
        properties.setProperty(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.StringSerde.class.getName());

        // create a topology
        StreamsBuilder streamsBuilder = new StreamsBuilder();

        // input topic
        KStream<String,String> inputTopic = streamsBuilder.stream("twitter_tweets");
        KStream<String,String> filteredStream = inputTopic.filter(
                (k,jsonValue)->extractUserFollowersInTweets(jsonValue)>10000
                        // filter for tweets which has a user of over 10000 followers
        );
        filteredStream.to("important_tweets");

        // build topology
        KafkaStreams kafkaStreams = new KafkaStreams(streamsBuilder.build(), properties);

        // start our streams application
        kafkaStreams.start();

    }

    private static JsonParser jsonParser = new JsonParser();
    private static int extractUserFollowersInTweets(String tweetJson) {
        try {
            return jsonParser.parse(tweetJson)
                    .getAsJsonObject()
                    .get("user")
                    .getAsJsonObject()
                    .get("followers_count")
                    .getAsInt();
        } catch (NullPointerException e){
            return 0;
        }
    }

} 

当我尝试运行此代码时，出现此错误：

线程“main”中的异常 java.lang.NoClassDefFoundError: org/apache/kafka/common/metrics/MetricsContext at tutorial4.StreamsFilterTweet.main(StreamsFilterTweet.java:35) 由：java.lang.ClassNotFoundException: org.apache.kafka .common.metrics.MetricsContext 在 java.base/jdk.internal.loader.BuiltinClassLoader.loadClass(BuiltinClassLoader.java:581) 在 java.base/jdk.internal.loader.ClassLoaders$AppClassLoader.loadClass(ClassLoaders.java:178)在 java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:521) ... 1 更多

根据堆栈跟踪，这一行导致了问题：

KafkaStreams kafkaStreams = new KafkaStreams(streamsBuilder.build(), properties);

我正在使用 Maven 并具有以下依赖项：

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.0.0</version>
</dependency>

<dependency>
    <groupId>org.slf4j</groupId>
    <artifactId>slf4j-simple</artifactId>
    <version>1.7.25</version>
</dependency>

<dependency>
    <groupId>org.apache.maven.plugins</groupId>
    <artifactId>maven-compiler-plugin</artifactId>
    <version>3.8.1</version>
</dependency>

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-streams</artifactId>
    <version>2.6.0</version>
</dependency>

<dependency>
    <groupId>com.google.code.gson</groupId>
    <artifactId>gson</artifactId>
    <version>2.8.6</version>
</dependency>

我找不到为什么会发生这个问题。我怎么解决这个问题？

Answer 1

问题不在于kafka-streams. 您在依赖项kafka-streams和kafka-clients. 所以你需要升级kafka-clients到2.6.0版本或降级kafka-streams到2.0.0版本。

Kafka Streams 依赖于kafka-clients，因此如果您将其中两个声明为不同版本的依赖项，您将面临此类错误。

最后，正如我在您的代码中看到的，您可以kafka-clients从POM.