Search code examples
apache-storm

Storm-Signals giving error


Hi I am running a topology and using storm signal to send message to topology again and again. But I am getting error in logs file.

ClassDefNotFound: backtype.storm.contrib.signals.spout.BaseSignalSpout

Can any one tell me the reason why its unable to find the class?

UPDATE: CODE FILES UPDATED

My Spout class:

@SuppressWarnings("serial")
public class NumberSpout extends BaseSignalSpout
{
    public NumberSpout(String name) {
        super(name);
        // TODO Auto-generated constructor stub
    }

    private static final Logger LOGGER = Logger.getLogger(NumberSpout.class);
    @SuppressWarnings("rawtypes")
    @Override
    public void open( Map conf, TopologyContext context, SpoutOutputCollector collector ) 
    {   super.open(conf, context, collector);
        LOGGER.info("In open method of spout");
    }

    @Override
    public void nextTuple() 
    {
    }

    @Override
    public void ack(Object id) 
    {
    }

    @Override
    public void fail(Object id) 
    {
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) 
    {
        declarer.declare( new Fields( "name" ) );
    }

    @Override
    public void onSignal(byte[] data) {
        LOGGER.info("Received signal: " + new String(data));
    }
}

My Toplogy Class:

public class PrimeNumberTopology 
{
    private static final Logger LOGGER = Logger.getLogger(PrimeNumberTopology.class);

public static void load(){
    TopologyBuilder builder = new TopologyBuilder();
    builder.setSpout( "signal-spout", new NumberSpout("test-signal-spout"));
    Config conf = new Config();
    conf.put(Config.NIMBUS_HOST, "127.0.0.1");
    conf.setDebug(true);
    Map storm_conf = Utils.readStormConfig();
    storm_conf.put("nimbus.host", "127.0.0.1");
    Client client = NimbusClient.getConfiguredClient(storm_conf)
            .getClient();
    String inputJar = "/home/jamil/Downloads/storm-twitter-word-count-master/target/storm-test-1.0-SNAPSHOT.jar";
    NimbusClient nimbus = new NimbusClient("127.0.0.1",6627);
    // upload topology jar to Cluster using StormSubmitter
    String uploadedJarLocation = StormSubmitter.submitJar(storm_conf,
            inputJar);
    try {
        String jsonConf = JSONValue.toJSONString(storm_conf);
        nimbus.getClient().submitTopology("test1topology",
                uploadedJarLocation, jsonConf, builder.createTopology());
    } catch (AlreadyAliveException ae) {
        ae.printStackTrace();
    } catch (InvalidTopologyException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    } catch (TException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    }

}
public static void main(String[] args) 
{   
    load();
    /*SignalClient sc = new SignalClient("localhost:2181", "test-signal-spout");
    sc.start();
    try {
        sc.send("Hello Signal Spout!".getBytes());
    } catch (Exception e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    } finally {
        sc.close();
    }*/

}
}

My Pom.xml File

<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.storm</groupId>
    <artifactId>storm-test</artifactId>
    <version>1.0-SNAPSHOT</version>
    <packaging>jar</packaging>

    <name>storm-test</name>
    <url>http://maven.apache.org</url>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    </properties>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>2.0.2</version>
                <configuration>
                    <source>1.6</source>
                    <target>1.6</target>
                </configuration>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-jar-plugin</artifactId>
                <configuration>
                    <archive>
                        <manifest>
                            <addClasspath>true</addClasspath>
                            <classpathPrefix>lib/</classpathPrefix>
                            <mainClass>com.storm.PrimeNumberTopology</mainClass>
                        </manifest>
                    </archive>
                </configuration>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-dependency-plugin</artifactId>

            </plugin>
        </plugins>
    </build>

    <dependencies>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>3.8.1</version>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>com.github.ptgoetz</groupId>
            <artifactId>storm-signals</artifactId>
            <version>0.2.0</version>
        </dependency>
        <dependency>
            <groupId>storm</groupId>
            <artifactId>storm-lib</artifactId>
            <version>0.8.1</version>

            <!-- keep storm out of the jar-with-dependencies <scope>provided</scope> -->
        </dependency>
    </dependencies>

    <repositories>
        <repository>
            <id>github-releases</id>
            <url>http://oss.sonatype.org/content/repositories/github-releases/</url>
        </repository>
        <repository>
            <id>clojars.org</id>
            <url>http://clojars.org/repo</url>
        </repository>
    </repositories>
</project>

My Error Log File:

2014-10-26 19:45:19 b.s.d.worker [ERROR] Error on initialization of server mk-worker
java.lang.NoClassDefFoundError: backtype/storm/contrib/signals/spout/BaseSignalSpout
    at java.lang.ClassLoader.defineClass1(Native Method) ~[na:1.6.0_32]
    at java.lang.ClassLoader.defineClass(ClassLoader.java:643) ~[na:1.6.0_32]
    at java.security.SecureClassLoader.defineClass(SecureClassLoader.java:142) ~[na:1.6.0_32]
    at java.net.URLClassLoader.defineClass(URLClassLoader.java:277) ~[na:1.6.0_32]
    at java.net.URLClassLoader.access$000(URLClassLoader.java:73) ~[na:1.6.0_32]
    at java.net.URLClassLoader$1.run(URLClassLoader.java:212) ~[na:1.6.0_32]
    at java.security.AccessController.doPrivileged(Native Method) ~[na:1.6.0_32]
    at java.net.URLClassLoader.findClass(URLClassLoader.java:205) ~[na:1.6.0_32]
    at java.lang.ClassLoader.loadClass(ClassLoader.java:323) ~[na:1.6.0_32]
    at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:294) ~[na:1.6.0_32]
    at java.lang.ClassLoader.loadClass(ClassLoader.java:268) ~[na:1.6.0_32]
    at java.lang.Class.forName0(Native Method) ~[na:1.6.0_32]
    at java.lang.Class.forName(Class.java:270) ~[na:1.6.0_32]
    at java.io.ObjectInputStream.resolveClass(ObjectInputStream.java:624) ~[na:1.6.0_32]
    at java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:1611) ~[na:1.6.0_32]
    at java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1516) ~[na:1.6.0_32]
    at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1770) ~[na:1.6.0_32]
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1349) ~[na:1.6.0_32]
    at java.io.ObjectInputStream.readObject(ObjectInputStream.java:369) ~[na:1.6.0_32]
    at backtype.storm.utils.Utils.deserialize(Utils.java:82) ~[storm-core-0.9.1-incubating.jar:0.9.1-incubating]
    at backtype.storm.utils.Utils.getSetComponentObject(Utils.java:218) ~[storm-core-0.9.1-incubating.jar:0.9.1-incubating]
    at backtype.storm.daemon.task$get_task_object.invoke(task.clj:73) ~[storm-core-0.9.1-incubating.jar:0.9.1-incubating]
    at backtype.storm.daemon.task$mk_task_data$fn__3402.invoke(task.clj:180) ~[na:na]
    at backtype.storm.util$assoc_apply_self.invoke(util.clj:792) ~[storm-core-0.9.1-incubating.jar:0.9.1-incubating]
    at backtype.storm.daemon.task$mk_task_data.invoke(task.clj:173) ~[storm-core-0.9.1-incubating.jar:0.9.1-incubating]
    at backtype.storm.daemon.task$mk_task.invoke(task.clj:184) ~[storm-core-0.9.1-incubating.jar:0.9.1-incubating]
    at backtype.storm.daemon.executor$mk_executor$fn__3785.invoke(executor.clj:320) ~[na:na]
    at clojure.core$map$fn__4087.invoke(core.clj:2432) ~[clojure-1.4.0.jar:na]
    at clojure.lang.LazySeq.sval(LazySeq.java:42) ~[clojure-1.4.0.jar:na]
    at clojure.lang.LazySeq.seq(LazySeq.java:60) ~[clojure-1.4.0.jar:na]
    at clojure.lang.RT.seq(RT.java:473) ~[clojure-1.4.0.jar:na]
    at clojure.core$seq.invoke(core.clj:133) ~[clojure-1.4.0.jar:na]
    at clojure.core.protocols$seq_reduce.invoke(protocols.clj:30) ~[clojure-1.4.0.jar:na]
    at clojure.core.protocols$fn__5875.invoke(protocols.clj:54) ~[clojure-1.4.0.jar:na]
    at clojure.core.protocols$fn__5828$G__5823__5841.invoke(protocols.clj:13) ~[clojure-1.4.0.jar:na]
    at clojure.core$reduce.invoke(core.clj:6030) ~[clojure-1.4.0.jar:na]
    at clojure.core$into.invoke(core.clj:6077) ~[clojure-1.4.0.jar:na]
    at backtype.storm.daemon.executor$mk_executor.invoke(executor.clj:320) ~[storm-core-0.9.1-incubating.jar:0.9.1-incubating]
    at backtype.storm.daemon.worker$eval4413$exec_fn__1102__auto____4414$iter__4419__4423$fn__4424.invoke(worker.clj:375) ~[na:na]
    at clojure.lang.LazySeq.sval(LazySeq.java:42) ~[clojure-1.4.0.jar:na]
    at clojure.lang.LazySeq.seq(LazySeq.java:60) ~[clojure-1.4.0.jar:na]
    at clojure.lang.RT.seq(RT.java:473) ~[clojure-1.4.0.jar:na]
    at clojure.core$seq.invoke(core.clj:133) ~[clojure-1.4.0.jar:na]
    at clojure.core$dorun.invoke(core.clj:2725) ~[clojure-1.4.0.jar:na]
    at clojure.core$doall.invoke(core.clj:2741) ~[clojure-1.4.0.jar:na]
    at backtype.storm.daemon.worker$eval4413$exec_fn__1102__auto____4414.invoke(worker.clj:375) ~[na:na]
    at clojure.lang.AFn.applyToHelper(AFn.java:185) ~[clojure-1.4.0.jar:na]
    at clojure.lang.AFn.applyTo(AFn.java:151) ~[clojure-1.4.0.jar:na]
    at clojure.core$apply.invoke(core.clj:601) ~[clojure-1.4.0.jar:na]
    at backtype.storm.daemon.worker$eval4413$mk_worker__4469.doInvoke(worker.clj:344) ~[na:na]
    at clojure.lang.RestFn.invoke(RestFn.java:512) ~[clojure-1.4.0.jar:na]
    at backtype.storm.daemon.worker$_main.invoke(worker.clj:454) ~[storm-core-0.9.1-incubating.jar:0.9.1-incubating]
    at clojure.lang.AFn.applyToHelper(AFn.java:172) ~[clojure-1.4.0.jar:na]
    at clojure.lang.AFn.applyTo(AFn.java:151) ~[clojure-1.4.0.jar:na]
    at backtype.storm.daemon.worker.main(Unknown Source) ~[storm-core-0.9.1-incubating.jar:0.9.1-incubating]
Caused by: java.lang.ClassNotFoundException: backtype.storm.contrib.signals.spout.BaseSignalSpout
    at java.net.URLClassLoader$1.run(URLClassLoader.java:217) ~[na:1.6.0_32]
    at java.security.AccessController.doPrivileged(Native Method) ~[na:1.6.0_32]
    at java.net.URLClassLoader.findClass(URLClassLoader.java:205) ~[na:1.6.0_32]
    at java.lang.ClassLoader.loadClass(ClassLoader.java:323) ~[na:1.6.0_32]
    at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:294) ~[na:1.6.0_32]
    at java.lang.ClassLoader.loadClass(ClassLoader.java:268) ~[na:1.6.0_32]
    ... 55 common frames omitted
2014-10-26 19:45:19 b.s.util [INFO] Halting process: ("Error on initialization")

After GAS Updation New Error:

java.lang.RuntimeException: Fail to construct messaging plugin from plugin backtype.storm.messaging.zmq
    at backtype.storm.messaging.TransportFactory.makeContext(TransportFactory.java:53) ~[storm-core-0.9.1-incubating.jar:0.9.1-incubating]
    at backtype.storm.daemon.worker$worker_data$fn__4263.invoke(worker.clj:185) ~[na:na]
    at backtype.storm.util$assoc_apply_self.invoke(util.clj:792) ~[storm-core-0.9.1-incubating.jar:0.9.1-incubating]
    at backtype.storm.daemon.worker$worker_data.invoke(worker.clj:181) ~[storm-core-0.9.1-incubating.jar:0.9.1-incubating]
    at backtype.storm.daemon.worker$eval4413$exec_fn__1102__auto____4414.invoke(worker.clj:353) ~[na:na]
    at clojure.lang.AFn.applyToHelper(AFn.java:185) ~[clojure-1.4.0.jar:na]
    at clojure.lang.AFn.applyTo(AFn.java:151) ~[clojure-1.4.0.jar:na]
    at clojure.core$apply.invoke(core.clj:601) ~[clojure-1.4.0.jar:na]
    at backtype.storm.daemon.worker$eval4413$mk_worker__4469.doInvoke(worker.clj:344) ~[na:na]
    at clojure.lang.RestFn.invoke(RestFn.java:512) ~[clojure-1.4.0.jar:na]
    at backtype.storm.daemon.worker$_main.invoke(worker.clj:454) ~[storm-core-0.9.1-incubating.jar:0.9.1-incubating]
    at clojure.lang.AFn.applyToHelper(AFn.java:172) ~[clojure-1.4.0.jar:na]
    at clojure.lang.AFn.applyTo(AFn.java:151) ~[clojure-1.4.0.jar:na]
    at backtype.storm.daemon.worker.main(Unknown Source) ~[storm-core-0.9.1-incubating.jar:0.9.1-incubating]
Caused by: java.lang.ClassNotFoundException: backtype.storm.messaging.zmq
    at java.net.URLClassLoader$1.run(URLClassLoader.java:217) ~[na:1.6.0_32]
    at java.security.AccessController.doPrivileged(Native Method) ~[na:1.6.0_32]
    at java.net.URLClassLoader.findClass(URLClassLoader.java:205) ~[na:1.6.0_32]
    at java.lang.ClassLoader.loadClass(ClassLoader.java:323) ~[na:1.6.0_32]
    at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:294) ~[na:1.6.0_32]
    at java.lang.ClassLoader.loadClass(ClassLoader.java:268) ~[na:1.6.0_32]
    at java.lang.Class.forName0(Native Method) ~[na:1.6.0_32]
    at java.lang.Class.forName(Class.java:190) ~[na:1.6.0_32]
    at backtype.storm.messaging.TransportFactory.makeContext(TransportFactory.java:38) ~[storm-core-0.9.1-incubating.jar:0.9.1-incubating]
    ... 13 common frames omitted
2014-10-27 02:19:06 b.s.util [INFO] Halting process: ("Error on initialization")

Solution

  • I tried you code,

    The problem is that the your jar cannot load the library, I added this on maven file (pom.xml) :

    <plugin>
                    <artifactId>maven-assembly-plugin</artifactId>
                    <configuration>
                        <archive>
                            <manifest>
                                <mainClass>PrimeNumberTopology</mainClass>
                            </manifest>
                        </archive>
                        <descriptorRefs>
                            <descriptorRef>jar-with-dependencies</descriptorRef>
                        </descriptorRefs>
                    </configuration>
                </plugin> 
    

    Then as you can read to the POM.xml, I removed storm lib from the jar-with-dependencies (when you deploy a jar you have to remove the storm-lib into your jar)

    <dependency>
                <groupId>storm</groupId>
                <artifactId>storm-lib</artifactId>
                <version>0.9.0</version>
                <scope>provided</scope> <---- added this,that exclude from the jar-with-dependencies
                <!-- keep storm out of the jar-with-dependencies <scope></scope> -->
            </dependency>
    

    In this way your jar contains only your code and the storm signal jar.

    I executed:

    storm jar /Projects/StackTest897/target/storm-test-1.0-SNAPSHOT-jar-with-dependencies.jar PrimeNumberTopology
    

    I updated the storm lib with the 0.9.0 because I'm using the last version, and your code does not compile here:

    NimbusClient("127.0.0.1",6627);
    

    Just fix it, or just use your environment and it should work correctly.

    Hope it helps.

    **EDIT****

    Yes, this is another kind of the problem, it depends on the version of Storm you're running.

    Please read Storm workers not starting

    or here:

    http://mail-archives.apache.org/mod_mbox/storm-user/201405.mbox/%3CCANvtHewkSfRJUzBPh74Nu8aUR7eoEh38r+KPgpwML5gZvoq7_g@mail.gmail.com%3E