1

Всем привет.
У меня есть приложение, которое работает с Stream Kafka.

Я в нем также пытаюсь создать топики но получаю ошибку:

INFO Terminate ZkClient event thread. (org.I0Itec.zkclient.ZkEventThread:83)
INFO Session: 0x1000000b299004d closed (org.apache.zookeeper.ZooKeeper:684)
INFO EventThread shut down for session: 0x1000000b299004d (org.apache.zookeeper.ClientCnxn:519)
Exception in thread "main" java.lang.NoClassDefFoundError: org/apache/kafka/common/protocol/SecurityProtocol
at kafka.server.Defaults$.(KafkaConfig.scala:120)
at kafka.server.Defaults$.(KafkaConfig.scala)
at kafka.log.Defaults$.(LogConfig.scala:29)
at kafka.log.Defaults$.(LogConfig.scala)
at kafka.log.LogConfig$.(LogConfig.scala:137)
at kafka.log.LogConfig$.(LogConfig.scala)
at kafka.admin.AdminUtils$.createOrUpdateTopicPartitionAssignmentPathInZK(AdminUtils.scala:268)
at kafka.admin.AdminUtils$.createTopic(AdminUtils.scala:237)
at kafka.admin.AdminUtils.createTopic(AdminUtils.scala)
at com.testStockStreams.PropertiesConfig.createTopic(PropertiesConfig.java:38)
at com.testStockStreams.StreamsApplication.main(StreamsApplication.java:23)
Caused by: java.lang.ClassNotFoundException: org.apache.kafka.common.protocol.SecurityProtocol
at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:349)
at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
... 11 more

Process finished with exit code 1

Ошибка возникает в строке класса PropertiesConfig во время попытки создать топик:

AdminUtils.createTopic(zkUtils, topicName, 10, 1, new Properties());

Подозреваю, что проблема в зависимостях, но найти решение никак не удается.

Буду очень благодарен за помощь. Код по тексту ниже:

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.testStockStreams</groupId>
    <artifactId>test-stock-streams</artifactId>
    <version>1.0-SNAPSHOT</version>
    <packaging>jar</packaging>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
        <java.version>1.8</java.version>
    </properties>

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

        <dependency>
            <groupId>com.101tec</groupId>
            <artifactId>zkclient</artifactId>
            <version>0.10</version>
        </dependency>

        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka_2.11</artifactId>
            <version>0.9.0.1</version>
        </dependency>

        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-streams-test-utils</artifactId>
            <version>1.1.0</version>
            <scope>test</scope>
        </dependency>

        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.12</version>
            <scope>test</scope>
        </dependency>

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

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.3</version>
                <inherited>true</inherited>
                <configuration>
                    <source>${java.version}</source>
                    <target>${java.version}</target>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

class PropertiesConfig

package com.testStockStreams;


import kafka.admin.AdminUtils;
import kafka.utils.ZKStringSerializer$;
import kafka.utils.ZkUtils;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.ZkConnection;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.errors.TopicExistsException;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.StreamsConfig;

import java.util.Properties;

public class PropertiesConfig {

    private final static String BOOTSTRAP_SERVER = "localhost:9092";
    private final static String ZK_SERVER = "localhost:2181";

    public static Properties initConfig() {
        Properties config = new Properties();
        config.put(StreamsConfig.APPLICATION_ID_CONFIG , "KafkaProducer");
        config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG , BOOTSTRAP_SERVER);
        config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG , "earliest");
        config.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG , Serdes.String().getClass());
        config.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG , Serdes.Double().getClass());

        return config;
    }

    public static void createTopic(String topicName) {
        ZkClient zkClient = new ZkClient(ZK_SERVER, 10000, 10000, ZKStringSerializer$.MODULE$);

        ZkUtils zkUtils = new ZkUtils(zkClient, new ZkConnection(ZK_SERVER), false);

        try {
            AdminUtils.createTopic(zkUtils, topicName, 10, 1, new Properties());
        } catch (TopicExistsException e) {
            System.out.println("*** Topic '" + topicName + "' already exist.");
        } finally {
            zkClient.close();
        }
    }
}

class StreamsApplication

package com.testStockStreams;

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.Topology;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;

import java.util.Properties;

public class StreamsApplication {

    private final static String TOPIC_MASTER = "stock-topic-master";
    private final static String TOPIC_LAST_DATA = "stock-topic-last-data";
    private final static String TOPIC_LESS = "stock-topic-less";
    private final static String TOPIC_MORE = "stock-topic-more";
    private final static String TOPIC_AVERAGES_STATISTICAL = "stock-topic-averages";


    public static void main(String[] args) {
        Properties config = PropertiesConfig.initConfig();
        PropertiesConfig.createTopic(TOPIC_LAST_DATA);
        PropertiesConfig.createTopic(TOPIC_LESS);
        PropertiesConfig.createTopic(TOPIC_MORE);
        PropertiesConfig.createTopic(TOPIC_AVERAGES_STATISTICAL);

        StreamsApplication streamsApp  = new StreamsApplication();
        KafkaStreams streams = new KafkaStreams(streamsApp.createTopology(), config);
        streams.start();

        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
    }

    public Topology createTopology() {
        StreamsBuilder builder = new StreamsBuilder();
        KStream<String, Double> quotes = builder.stream(TOPIC_MASTER);

        /* save latest data on quotes */
        KTable<String, Double> quotesSlice = quotes.groupByKey().reduce((previousValue, newValue) -> newValue);

        /* save average value of quotations in KTable */
        KTable<String, Long> countValues = quotes.groupByKey().count();
        KTable<String, Double> sumValues = quotes.groupByKey().reduce((v1, v2) -> v1 + v2);
        KTable<String, Double> averages = sumValues.join(countValues, (sum, count) -> sum/count.doubleValue());

        /* tables (1) and (2) need to be additionally merged into two topics in Kafka */
        averages.to(Serdes.String(), Serdes.Double(), TOPIC_AVERAGES_STATISTICAL);
        quotesSlice.to(Serdes.String(), Serdes.Double(), TOPIC_LAST_DATA);

        /* split the input stream of quotations into the other two and save them in two different topics,
           one where the value of the quotes> = 2.5, in another <2.5 */
        KStream<String, Double> valuesLess = quotes.filter((key, value) -> value < 2.5);
        valuesLess.to(Serdes.String(), Serdes.Double(), TOPIC_LESS);
        KStream<String, Double> valuesMore = quotes.filter((key, value) -> value >= 2.5);
        valuesMore.to(Serdes.String(), Serdes.Double(), TOPIC_MORE);

        return builder.build();
    }

}

1 ответ 1

0

Попробуйте kafka_2.12 вместо kafka_2.11. Или возможно, что надо поменять <version>. Проблема в том, что SecurityProtocol передвинули в другой пакет (из org.apache.kafka.common.protocol в org.apache.kafka.common.security.auth) и если использовать несочетающиеся версии - класс грузится из неправильного пакета. Мне помогли вот эти списки версий:

Ваш ответ

By clicking “Отправить ответ”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Всё ещё ищете ответ? Посмотрите другие вопросы с метками или задайте свой вопрос.