KafkaSpout BrokerHosts error

Question

I am trying to compile and run the storm-kafka-starter project at https://github.com/TheHydroImpulse/storm-kafka-starter

The main function for KafkaTopology looks like:

public class KafkaTopology {

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

    List<String> hosts = new ArrayList<String>();
    hosts.add("localhost");
    SpoutConfig kafkaConf = new SpoutConfig(StaticHosts.fromHostString(hosts,1),
                                        "test-topic","/kafkastorm","discovery");
    kafkaConf.scheme = new SchemeAsMultiScheme(new StringScheme());
    kafkaConf.forceStartOffsetTime(-2);
    TopologyBuilder builder = new TopologyBuilder();
    builder.setSpout("spout", kafkaSpout, 2);
    builder.setBolt("printer", new PrinterBolt()).shuffleGrouping("spout");

    Config config = new Config();
    config.setDebug(true);

    if(args!=null && args.length > 0) {
      config.setNumWorkers(3);
      StormSubmitter.submitTopology(args[0], config, builder.createTopology());
    } 
    else {        
      config.setMaxTaskParallelism(3);
      LocalCluster cluster = new LocalCluster();
      cluster.submitTopology("kafka", config, builder.createTopology());

      Thread.sleep(10000);

      cluster.shutdown();
    }
  }
}

The jar compiles using maven. But on running the topology, I get the error:

Exception in thread "main" java.lang.NoClassDefFoundError:      
storm/kafka/KafkaConfig$BrokerHosts
at java.lang.Class.getDeclaredMethods0(Native Method)
at java.lang.Class.privateGetDeclaredMethods(Class.java:2451)
at java.lang.Class.getMethod0(Class.java:2694)
at java.lang.Class.getMethod(Class.java:1622) 
at sun.launcher.LauncherHelper.getMainMethod(LauncherHelper.java:494)
at sun.launcher.LauncherHelper.checkAndLoadMain(LauncherHelper.java:486)
Caused by: java.lang.ClassNotFoundException: storm.kafka.KafkaConfig$BrokerHosts
at java.net.URLClassLoader$1.run(URLClassLoader.java:366)
at java.net.URLClassLoader$1.run(URLClassLoader.java:355)
at java.security.AccessController.doPrivileged(Native Method)
at java.net.URLClassLoader.findClass(URLClassLoader.java:354)
at java.lang.ClassLoader.loadClass(ClassLoader.java:423)
at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:308)
at java.lang.ClassLoader.loadClass(ClassLoader.java:356)
... 6 more

The local repository does have BrokerHosts in the storm-kafka jar and I have imported the KafkaConfig library in my java file. I cannot figure out the cause of the error. Any suggestions would be appreciated.

Answer 1

I went through the storm-user groups and the issues on the storm-kafka-starter github page. It turns out that the error is due to two reasons:

1. Version incompatibility among versions of storm, kafka and kafka-storm
2. Missing jars on the classpath

My initial setup did not work even after including all the necessary dependency jars in the /storm/lib folder. Turns out that the storm-kafka-starter project mentioned works only for storm 0.9.x versions.

Also see post here on which setup works best - https://groups.google.com/d/msg/storm-user/V_j_JZmFsb4/E4_II9ork3UJ

Answer 2

I had similar issues using the 0.9.2_incubating version of Apache Storm.

The isssue is caused because the actual storm distribution doesn't have the kafka libraries in the /lib folder. I was able to resolve the error by copying the following libraries (that I used to compile & build the topology) to the /lib folder from where i ran Storm

• storm-kafka-0.9.2-incubating.jar
• kafka_2.10-0.8.1.1.jar
• scala-library-2.10.1.jar

Remember that the actual versions in your case might vary. Take the ones that you use to build your storm topology (i.e. from .m2 or .gradle dependencies folder)

Note: I am not using the exact same starter project mentioned above but the fix will be similar.

Answer 3

I went through the same integration woes. Finally got a working example together.

You are welcome to check it out here > https://github.com/buildlackey/cep

(click on the storm+kafka directory for a sample program that should get you up and running).

Answer 4

I failed to use to correct maven build command. Using

mvn clean install assembly:assembly

and using the jar with dependencies fixed it form me.

Answer 5

Changed pom.xml from

<dependency>
   <groupId>org.apache.storm</groupId>
   <artifactId>storm-kafka</artifactId>
   <version>1.0.2</version>
   <scope>provided</scope>
</dependency>

to

 <dependency>
   <groupId>org.apache.storm</groupId>
   <artifactId>storm-kafka</artifactId>
   <version>1.0.2</version>
</dependency>

Now it works for me!