Storm-Signals giving error

Question

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")

Are you using remote cluster or local ? could you post the code? — Gabriele Santomaggio, Oct 26 '14 at 14:45
Yes i am using the storm cluster on my machine. I will be updating the code in actual question. — Mr37037, Oct 26 '14 at 14:47

score 3 · Accepted Answer · edited May 23 '17 at 10:30

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

Hi Gas I resolved that error by your new instructions. But now its giving another error. Like error in initialization of server mk worker. classdefNotfound: PrimeNumberBolt. Now its unable to find my Bolt class. Now this error is due to what reasons? — Mr37037, Oct 27 '14 at 12:12
Actually i copied spout,bolt and topology class and mentioned them in tomcat listener class. So whenever tomcat starts the topology should be submitted. But its giving above error in log files of storm. whereas nimbus,supervisor,ui log files were giving no error. The line when i submit topology gives this error in log files. Can you tell me the reason please? — Mr37037, Oct 27 '14 at 12:14
Ignore above comment I have updated the new question over here.So it will be easy for understanding. Can you please answer it? http://stackoverflow.com/questions/26587648/storm-topology-giving-error-after-integrating-with-tomcat — Mr37037, Oct 27 '14 at 12:32

Storm-Signals giving error

1 Answers1