NullPointerException when testing multithreaded app

[liu]

The app works fine in single thread, but when I try to test multithreading, it throws the following NullPointer exception:

java.lang.NullPointerException

        at scala.collection.JavaConversions$JListWrapper.length(JavaConversions.scala:616)

        at scala.collection.SeqLike$class.size(SeqLike.scala:100)

        at scala.collection.JavaConversions$JListWrapper.size(JavaConversions.scala:615)

        at scala.collection.TraversableOnce$class.toArray(TraversableOnce.scala:233)

        at scala.collection.JavaConversions$JListWrapper.toArray(JavaConversions.scala:615)

        at spark.ParallelCollection$.slice(ParallelCollection.scala:86)

        at spark.ParallelCollection.<init>(ParallelCollection.scala:35)

        at spark.SparkContext.parallelize(SparkContext.scala:195)

        at spark.api.java.JavaSparkContext.parallelize(JavaSparkContext.scala:68)

        at spark.api.java.JavaSparkContext.parallelize(JavaSparkContext.scala:73)

        at com.gdv.spark.tagcloud.TagCloudJob.runJob(TagCloudJob.java:52)

        at com.gdv.spark.tagcloud.TagCloudJob.runJob(TagCloudJob.java:34)

        at com.gdv.spark.tagcloud.TagCloudJobExecutor.execute(TagCloudJobExecutor.java:38)

        at com.gdv.spark.tagcloud.TagCloudJobExecutor.execute(TagCloudJobExecutor.java:18)

        at com.gdv.spark.tagcloud.TagCloudJobExecutor$ThreadTester.run(TagCloudJobExecutor.java:83)

        at java.lang.Thread.run(Thread.java:679)

java.lang.NullPointerException

        at scala.collection.JavaConversions$JListWrapper.length(JavaConversions.scala:616)

        at scala.collection.SeqLike$class.size(SeqLike.scala:100)

        at scala.collection.JavaConversions$JListWrapper.size(JavaConversions.scala:615)

        at scala.collection.TraversableOnce$class.toArray(TraversableOnce.scala:233)

        at scala.collection.JavaConversions$JListWrapper.toArray(JavaConversions.scala:615)

        at spark.ParallelCollection$.slice(ParallelCollection.scala:86)

        at spark.ParallelCollection.<init>(ParallelCollection.scala:35)

        at spark.SparkContext.parallelize(SparkContext.scala:195)

        at spark.api.java.JavaSparkContext.parallelize(JavaSparkContext.scala:68)

        at spark.api.java.JavaSparkContext.parallelize(JavaSparkContext.scala:73)

        at com.gdv.spark.tagcloud.TagCloudJob.runJob(TagCloudJob.java:52)

        at com.gdv.spark.tagcloud.TagCloudJob.runJob(TagCloudJob.java:34)

        at com.gdv.spark.tagcloud.TagCloudJobExecutor.execute(TagCloudJobExecutor.java:38)

        at com.gdv.spark.tagcloud.TagCloudJobExecutor.execute(TagCloudJobExecutor.java:18)

        at com.gdv.spark.tagcloud.TagCloudJobExecutor$ThreadTester.run(TagCloudJobExecutor.java:83)

        at java.lang.Thread.run(Thread.java:679)

I created a separate JavaSparkContext and a list of Strings for each thread.

Could someone point out where might be the issue?

Thank you

[Matei Zaharia]

You can't currently have two SparkContexts active at the same time in the same program. Reuse the same one in both threads. I believe you'll also have to do SparkEnv.set() in the second thread to set the SparkEnv object for the first one (obtained with SparkEnv.get()).

Matei

--
You received this message because you are subscribed to the Google Groups "Spark Users" group.
To unsubscribe from this group and stop receiving emails from it, send an email to spark-users...@googlegroups.com.
For more options, visit https://groups.google.com/groups/opt_out.
 
 

[liu]

Hi,

Thank you for the reply. If I reuse the same SparkContext, it throws Exception:

Caused by: java.net.BindException: Address already in use

        at sun.nio.ch.Net.bind(Native Method)

        at sun.nio.ch.ServerSocketChannelImpl.bind(ServerSocketChannelImpl.java:137)

        at sun.nio.ch.ServerSocketAdaptor.bind(ServerSocketAdaptor.java:77)

        at org.jboss.netty.channel.socket.nio.NioServerSocketPipelineSink.bind(NioServerSocketPipelineSink.java:131)

        at org.jboss.netty.channel.socket.nio.NioServerSocketPipelineSink.handleServerSocket(NioServerSocketPipelineSink.java:83)

        at org.jboss.netty.channel.socket.nio.NioServerSocketPipelineSink.eventSunk(NioServerSocketPipelineSink.java:57)

        at org.jboss.netty.channel.Channels.bind(Channels.java:569)

        at org.jboss.netty.channel.AbstractChannel.bind(AbstractChannel.java:186)

        at org.jboss.netty.bootstrap.ServerBootstrap$Binder.channelOpen(ServerBootstrap.java:343)

        at org.jboss.netty.channel.Channels.fireChannelOpen(Channels.java:170)

        at org.jboss.netty.channel.socket.nio.NioServerSocketChannel.<init>(NioServerSocketChannel.java:80)

        at org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory.newChannel(NioServerSocketChannelFactory.java:156)

        at org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory.newChannel(NioServerSocketChannelFactory.java:86)

        at org.jboss.netty.bootstrap.ServerBootstrap.bind(ServerBootstrap.java:277)

        ... 33 more

I found the following:

https://spark-project.atlassian.net/browse/SPARK-534

And I tried the following in each thread but not working:

SparkEnv.set(sc.env());

Thanks

[liu]

Hi,

According to the document, the SparkEnv is not thread safe, I don't understand how it works if get SparkEnv instance from the first one and set to it into the second thread which means the two threads share the same instance. 

Thanks

On Monday, February 11, 2013 11:58:46 AM UTC-8, Matei Zaharia wrote:

[liu]

Anyone shed some lights on this?

Thanks