CODEWITHSUNDEEP
javaapache-stormapache-kafka
Johnny
Johnny

Reputation: 7351

Unable to read data with StormSpout

I'm trying to read messages from a Kafka queue using KafkaSpout. I'm either getting nothing at all or the following error:

2 [Thread-10-kafka-storm-spout] ERROR util:0 - Async loop died!
java.lang.NoSuchMethodError: scala.Predef$.augmentString(Ljava/lang/String;)Ljava/lang/String;
    at kafka.consumer.SimpleConsumer.<init>(SimpleConsumer.scala:38)
    at kafka.javaapi.consumer.SimpleConsumer.<init>(SimpleConsumer.scala:34)
    at storm.kafka.DynamicPartitionConnections.register(DynamicPartitionConnections.java:43)
    at storm.kafka.PartitionManager.<init>(PartitionManager.java:57)
    at storm.kafka.ZkCoordinator.refresh(ZkCoordinator.java:80)
    at storm.kafka.ZkCoordinator.getMyManagedPartitions(ZkCoordinator.java:52)
    at storm.kafka.KafkaSpout.nextTuple(KafkaSpout.java:118)
    at backtype.storm.daemon.executor$fn__3284$fn__3299$fn__3328.invoke(executor.clj:563)
    at backtype.storm.util$async_loop$fn__452.invoke(util.clj:431)
    at clojure.lang.AFn.run(AFn.java:24)
    at java.lang.Thread.run(Thread.java:744)
11 [Thread-10-kafka-storm-spout] ERROR executor:0 - 
java.lang.NoSuchMethodError: scala.Predef$.augmentString(Ljava/lang/String;)Ljava/lang/String;
    at kafka.consumer.SimpleConsumer.<init>(SimpleConsumer.scala:38)
    at kafka.javaapi.consumer.SimpleConsumer.<init>(SimpleConsumer.scala:34)
    at storm.kafka.DynamicPartitionConnections.register(DynamicPartitionConnections.java:43)
    at storm.kafka.PartitionManager.<init>(PartitionManager.java:57)
    at storm.kafka.ZkCoordinator.refresh(ZkCoordinator.java:80)
    at storm.kafka.ZkCoordinator.getMyManagedPartitions(ZkCoordinator.java:52)
    at storm.kafka.KafkaSpout.nextTuple(KafkaSpout.java:118)
    at backtype.storm.daemon.executor$fn__3284$fn__3299$fn__3328.invoke(executor.clj:563)
    at backtype.storm.util$async_loop$fn__452.invoke(util.clj:431)
    at clojure.lang.AFn.run(AFn.java:24)
    at java.lang.Thread.run(Thread.java:744)

Here's my code:

TopologyBuilder builder = new TopologyBuilder();
        String TOPIC_NAME = "topic";
        String spoutName = "kafka-storm-spout";             

        BrokerHosts brokerHosts = new ZkHosts("localhost:2181");
        SpoutConfig kafkaConfig = new SpoutConfig(brokerHosts, TOPIC_NAME, "", "storm");

        builder.setSpout(spoutName, new KafkaSpout(kafkaConfig), 1);
        builder.setBolt("kafka-bolt", new TestBolt()).shuffleGrouping(spoutName);

        Config config = new Config();
        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("kafka-test", config, builder.createTopology());

        System.out.println("Topology submitted");
        Utils.sleep(5000);
        System.out.println("Shutting down");
        cluster.shutdown();

Any ideas?

Upvotes: 0

Views: 406

Answers (1)

Maxime
Maxime

Reputation: 465

You are most likely trying to use different versions of scala. Kafka is built for different versions of scala (https://kafka.apache.org/downloads.html). Look into your dependencies and make sure you're using only one version of scala.

Upvotes: 1

Related Questions