NoClassDefFoundError: kafka/api/OffsetRequest

Question

I am trying to write application for real time processing with apache storm , kafka and trident but in initialization of TridentKafkaConfig i see this error

Exception in thread "main" java.lang.NoClassDefFoundError: kafka/api/OffsetRequest
at storm.kafka.KafkaConfig.<init>(KafkaConfig.java:43)
at storm.kafka.trident.TridentKafkaConfig.<init>(TridentKafkaConfig.java:30)
at spout.TestSpout.<clinit>(TestSpout.java:22)
at IOTTridentTopology.initializeTridentTopology(IOTTridentTopology.java:31)
at IOTTridentTopology.main(IOTTridentTopology.java:26)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:601)
at com.intellij.rt.execution.application.AppMain.main(AppMain.java:144)
Caused by: java.lang.ClassNotFoundException: kafka.api.OffsetRequest
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)
... 10 more

my spout class is

public class TestSpout extends OpaqueTridentKafkaSpout {

private static TridentKafkaConfig config;
private static BrokerHosts HOSTS = new ZkHosts(TridentConfig.ZKHOSTS);
private static String TOPIC = "test";
private static int BUFFER_SIZE = TridentConfig.BUFFER_SIZE;

static{
    config = new TridentKafkaConfig(HOSTS, TOPIC);
    config.scheme = new SchemeAsMultiScheme(new RawScheme());
    config.bufferSizeBytes = BUFFER_SIZE;
}

public TestSpout(TridentKafkaConfig config) {
    super(config);
}

public TestSpout() {
    super(config);
}
}

main class:

 public static void main(String[] args) {
    initializeTridentTopology();
}

private static void initializeTridentTopology() {
    TridentTopology topology = new TridentTopology();
    TestSpout spout = new TestSpout();
    //////////////// test  //////////////////////

    topology.newStream("testspout", spout).each(spout.getOutputFields(), new TestFunction(), new Fields());

    ///////////////  end test ///////////////////

    LocalCluster cluster = new LocalCluster();

    Config config = new Config();
    config.setDebug(false);
    config.setMaxTaskParallelism(1);
    config.registerSerialization(storm.kafka.trident.GlobalPartitionInformation.class);
    config.registerSerialization(java.util.TreeMap.class);
    config.setNumWorkers(5);

    config.setFallBackOnJavaSerialization(true);

    cluster.submitTopology("KafkaTrident", config, topology.build());

}

and my pom.xml:

<?xml version="1.0" encoding="UTF-8"?>

http://maven.apache.org/xsd/maven-4.0.0.xsd"> 4.0.0

<groupId>IOT</groupId>
<artifactId>ver0.1</artifactId>
<version>1.0-SNAPSHOT</version>

<dependencies>


    <dependency>
        <groupId>org.apache.storm</groupId>
        <artifactId>storm-core</artifactId>
        <version>0.9.3</version>
    </dependency>


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



</dependencies>

I am trying different version of storm-kafka (0.9.3 and 0.9.4 and 0.9.5 and 0.9.6 and 0.10.0) and storm-core (9.3 and 9.4 and 9.6)

But I still see my previous error

by googling i found this link but ...

ClassNotFoundException: kafka.api.OffsetRequest

Make sure that `kafka/api/OffsetRequest.class` is contained in your jar file. — Matthias J. Sax, Jan 06 '16 at 07:24
NoClassDefFoundError can be also because you have different versions of kafka/api/OffsetRequest, Have a look at your classpath! — awsome, Jan 06 '16 at 08:12

score 4 · Accepted Answer · answered Jan 06 '16 at 11:01

after some googling i found this link https://github.com/wurstmeister/storm-kafka-0.8-plus-test and found my answer in pom.xml file

by adding this code and find compatible version of kafka all problem resolved

 <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka_2.11</artifactId>
        <version>0.9.0.0</version>
        <exclusions>
            <exclusion>
                <groupId>org.apache.zookeeper</groupId>
                <artifactId>zookeeper</artifactId>
            </exclusion>
            <exclusion>
                <groupId>org.slf4j</groupId>
                <artifactId>slf4j-log4j12</artifactId>
            </exclusion>
            <exclusion>
                <groupId>log4j</groupId>
                <artifactId>log4j</artifactId>
            </exclusion>
        </exclusions>
    </dependency>

score 2 · Answer 2 · answered Jan 06 '16 at 08:07

If you use LocalCluster deployment a storm topology you need to add the Kafka lib to your dependencies (for Storm 0.10.0):

    <dependency>
       <groupId>org.apache.kafka</groupId>
       <artifactId>kafka_2.9.2</artifactId>
       <version>0.8.1.1</version>
     </dependency>

kafka.api.OffsetRequest class is missed beacause org.apache.kafka is provided dependency for the storm-kafka: http://mvnrepository.com/artifact/org.apache.storm/storm-kafka/0.10.0. Please, see the Provided Dependencies section for details.

Tnx For Your Answer after adding this version all Problem was solved but i can't receive any thing in BaseFunction — Ali Kianinejad, Jan 06 '16 at 10:59

NoClassDefFoundError: kafka/api/OffsetRequest

2 Answers2