Get Storm Runo на Кафке - PullRequest
Новая
       19

Get Storm Runo на Кафке

0 голосов
/ 09 июня 2019

У меня проблема с выполнением образца для подключения шторма к Кафке. У меня кафка работает успешно. Следующим шагом будет подключение шторма.

Это пример кода. 

/**
 * @author Amit Kumar
 */
public class Topology {

    private static final Logger LOG = Logger.getLogger(Topology.class);

    public static void main(String[] args) {

        // Build Spout configuration using input command line parameters
        final BrokerHosts zkrHosts = new ZkHosts("gitserver:2181");
        final String kafkaTopic = "test";
        final String zkRoot = "";
        final String clientId = "client1";
        final SpoutConfig kafkaConf = new SpoutConfig(zkrHosts, kafkaTopic, zkRoot, clientId);
        kafkaConf.scheme = new SchemeAsMultiScheme(new StringScheme());

        // Build topology to consume message from kafka and print them on console
        final TopologyBuilder topologyBuilder = new TopologyBuilder();
        // Create KafkaSpout instance using Kafka configuration and add it to topology
        topologyBuilder.setSpout("kafka-spout", new KafkaSpout(kafkaConf), 1);
        //Route the output of Kafka Spout to Logger bolt to log messages consumed from Kafka
        topologyBuilder.setBolt("print-messages", new LoggerBolt()).globalGrouping("kafka-spout");

        // Submit topology to local cluster i.e. embedded storm instance in eclipse
        final LocalCluster localCluster = new LocalCluster();
        localCluster.submitTopology("kafka-topology", new HashMap(), topologyBuilder.createTopology());
    }
}

Я получаю ошибку, которую не понимаю. Кто может помочь? Я думаю, это связано со значением zkRoot. Я пробовал разные пути (например, установку zookeeper или данные, но ничего не работает. 

21790 [Thread-18-kafka-spout-executor[2 2]] INFO  o.a.s.k.ZkCoordinator - Task [1/1], Task-ID: 2 New partition managers: [Partition{host=gitserver:9092, topic=test, partition=0}]
21839 [Thread-18-kafka-spout-executor[2 2]] INFO  o.a.s.k.PartitionManager - Read partition information from: /client1/partition_0  --> null
21877 [Thread-18-kafka-spout-executor[2 2]] INFO  k.c.SimpleConsumer - Reconnect due to socket error: java.lang.NoSuchMethodError: org.apache.kafka.common.network.NetworkSend: method <init>(Ljava/lang/String;[Ljava/nio/ByteBuffer;)V not found
21879 [Thread-18-kafka-spout-executor[2 2]] ERROR o.a.s.util - Async loop died!
java.lang.NoSuchMethodError: org.apache.kafka.common.network.NetworkSend: method <init>(Ljava/lang/String;[Ljava/nio/ByteBuffer;)V not found
    at kafka.network.RequestOrResponseSend.<init>(RequestOrResponseSend.scala:41) ~[kafka_2.11-0.9.0.0.jar:?]
    at kafka.network.RequestOrResponseSend.<init>(RequestOrResponseSend.scala:44) ~[kafka_2.11-0.9.0.0.jar:?]
    at kafka.network.BlockingChannel.send(BlockingChannel.scala:112) ~[kafka_2.11-0.9.0.0.jar:?]
    at kafka.consumer.SimpleConsumer.liftedTree1$1(SimpleConsumer.scala:98) ~[kafka_2.11-0.9.0.0.jar:?]
    at kafka.consumer.SimpleConsumer.kafka$consumer$SimpleConsumer$$sendRequest(SimpleConsumer.scala:83) ~[kafka_2.11-0.9.0.0.jar:?]
    at kafka.consumer.SimpleConsumer.getOffsetsBefore(SimpleConsumer.scala:149) ~[kafka_2.11-0.9.0.0.jar:?]
    at kafka.javaapi.consumer.SimpleConsumer.getOffsetsBefore(SimpleConsumer.scala:79) ~[kafka_2.11-0.9.0.0.jar:?]
    at org.apache.storm.kafka.KafkaUtils.getOffset(KafkaUtils.java:81) ~[storm-kafka-1.2.2.jar:1.2.2]
    at org.apache.storm.kafka.KafkaUtils.getOffset(KafkaUtils.java:71) ~[storm-kafka-1.2.2.jar:1.2.2]
    at org.apache.storm.kafka.PartitionManager.<init>(PartitionManager.java:135) ~[storm-kafka-1.2.2.jar:1.2.2]
    at org.apache.storm.kafka.ZkCoordinator.refresh(ZkCoordinator.java:110) ~[storm-kafka-1.2.2.jar:1.2.2]
    at org.apache.storm.kafka.ZkCoordinator.getMyManagedPartitions(ZkCoordinator.java:71) ~[storm-kafka-1.2.2.jar:1.2.2]
    at org.apache.storm.kafka.KafkaSpout.nextTuple(KafkaSpout.java:135) ~[storm-kafka-1.2.2.jar:1.2.2]
    at org.apache.storm.daemon.executor$fn__10727$fn__10742$fn__10773.invoke(executor.clj:654) ~[storm-core-1.2.2.jar:1.2.2]
    at org.apache.storm.util$async_loop$fn__553.invoke(util.clj:484) [storm-core-1.2.2.jar:1.2.2]
    at clojure.lang.AFn.run(AFn.java:22) [clojure-1.7.0.jar:?]
    at java.lang.Thread.run(Thread.java:745) [?:1.8.0_73]
21884 [Thread-18-kafka-spout-executor[2 2]] ERROR o.a.s.d.executor -

1 Ответ

1 голос
/ 09 июня 2019

Пожалуйста, используйте storm-kafka-client вместо storm-kafka для нового кода, последний не рекомендуется.

Вероятно, вы видите ошибку, вызванную смешиванием разных версий Kafka в вашем пути к классам.Я посмотрел класс, из которого вы получаете ошибку, и конструктор, в котором упоминается ошибка, присутствует в Kafka 0,9 (https://github.com/apache/kafka/blob/0.9.0.0/clients/src/main/java/org/apache/kafka/common/network/NetworkSend.java#L26), но не 0,8 (https://github.com/apache/kafka/blob/0.8.2.2/clients/src/main/java/org/apache/kafka/common/network/NetworkSend.java#L26)

) Скорее всего, вы случайносмешивание классов Kafka 0.8 в вашей сборке. Я бы использовал mvn dependency:tree, чтобы проверить вашу топологию и убедиться, что все фляги Kafka принадлежат к одной и той же версии.

...