The issue is related to this https://issues.apache.org/jira/browse/SPARK-13906
.set("spark.rpc.netty.dispatcher.numThreads","2") seem to fix the problem On Tue, Mar 15, 2016 at 6:45 AM, David Gomez Saavedra <mikr...@gmail.com> wrote: > I have updated the config since I realized the actor system was listening > on driver port + 1. So changed the ports in my program + the docker images > > val conf = new SparkConf() > .setMaster(sparkMaster) > //.setMaster("local[2]") > .setAppName(sparkApp) > .set("spark.cassandra.connection.host", CassandraConfig.host) > .set("spark.logConf", "true") > .set("spark.driver.port","7001") > .set("spark.driver.host","192.168.33.10") > .set("spark.fileserver.port","6002") > .set("spark.broadcast.port","6003") > .set("spark.replClassServer.port","6004") > .set("spark.blockManager.port","6005") > .set("spark.executor.port","6006") > > .set("spark.broadcast.factory","org.apache.spark.broadcast.HttpBroadcastFactory") > .setJars(sparkJars) > > Netstat of my stream app > > tcp6 0 0 :::6002 :::* LISTEN > 9314/java > tcp6 0 0 :::6003 :::* LISTEN > 9314/java > tcp6 0 0 :::6005 :::* LISTEN > 9314/java > tcp6 0 0 192.168.33.10:7001 :::* > LISTEN 9314/java > tcp6 0 0 192.168.33.10:7002 :::* > LISTEN 9314/java > tcp6 0 0 :::4040 :::* LISTEN > 9314/java > > netstat of the master running on docker > > Proto Recv-Q Send-Q Local Address Foreign Address State > PID/Program name > tcp6 0 0 172.18.0.3:7077 :::* > LISTEN - > tcp6 0 0 :::8080 :::* LISTEN > - > tcp6 0 0 172.18.0.3:6066 :::* > LISTEN - > > netstat of worker running on docker > > Proto Recv-Q Send-Q Local Address Foreign Address State > PID/Program name > tcp6 0 0 :::8081 :::* LISTEN > - > tcp6 0 0 :::6005 :::* LISTEN > - > tcp6 0 0 172.18.0.2:6006 :::* > LISTEN - > tcp6 0 0 172.18.0.2:8888 :::* > LISTEN - > > > so far still no success > > > > > > > > > > > > > > > > > On Mon, Mar 14, 2016 at 11:14 PM, Shixiong(Ryan) Zhu < > shixi...@databricks.com> wrote: > >> Could you use netstat to show the ports that the driver is listening? >> >> On Mon, Mar 14, 2016 at 1:45 PM, David Gomez Saavedra <mikr...@gmail.com> >> wrote: >> >>> hi everyone, >>> >>> I'm trying to set up spark streaming using akka with a similar example >>> of the word count provided. When using spark master in local mode >>> everything works but when I try to run it the driver and executors using >>> docker I get the following exception >>> >>> >>> 16/03/14 20:32:03 WARN NettyRpcEndpointRef: Error sending message [message >>> = Heartbeat(0,[Lscala.Tuple2;@5ad3f40c,BlockManagerId(0, 172.18.0.4, >>> 7005))] in 1 attempts >>> org.apache.spark.rpc.RpcTimeoutException: Cannot receive any reply in 10 >>> seconds. This timeout is controlled by spark.executor.heartbeatInterval >>> at >>> org.apache.spark.rpc.RpcTimeout.org$apache$spark$rpc$RpcTimeout$$createRpcTimeoutException(RpcTimeout.scala:48) >>> at >>> org.apache.spark.rpc.RpcTimeout$$anonfun$addMessageIfTimeout$1.applyOrElse(RpcTimeout.scala:63) >>> at >>> org.apache.spark.rpc.RpcTimeout$$anonfun$addMessageIfTimeout$1.applyOrElse(RpcTimeout.scala:59) >>> at >>> scala.runtime.AbstractPartialFunction.apply(AbstractPartialFunction.scala:36) >>> at scala.util.Failure$$anonfun$recover$1.apply(Try.scala:216) >>> at scala.util.Try$.apply(Try.scala:192) >>> at scala.util.Failure.recover(Try.scala:216) >>> at scala.concurrent.Future$$anonfun$recover$1.apply(Future.scala:324) >>> at scala.concurrent.Future$$anonfun$recover$1.apply(Future.scala:324) >>> at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:32) >>> at >>> org.spark-project.guava.util.concurrent.MoreExecutors$SameThreadExecutorService.execute(MoreExecutors.java:293) >>> at >>> scala.concurrent.impl.ExecutionContextImpl$$anon$1.execute(ExecutionContextImpl.scala:136) >>> at >>> scala.concurrent.impl.CallbackRunnable.executeWithValue(Promise.scala:40) >>> at >>> scala.concurrent.impl.Promise$DefaultPromise.tryComplete(Promise.scala:248) >>> at scala.concurrent.Promise$class.complete(Promise.scala:55) >>> at >>> scala.concurrent.impl.Promise$DefaultPromise.complete(Promise.scala:153) >>> at scala.concurrent.Future$$anonfun$map$1.apply(Future.scala:235) >>> at scala.concurrent.Future$$anonfun$map$1.apply(Future.scala:235) >>> at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:32) >>> at >>> scala.concurrent.BatchingExecutor$Batch$$anonfun$run$1.processBatch$1(BatchingExecutor.scala:63) >>> at >>> scala.concurrent.BatchingExecutor$Batch$$anonfun$run$1.apply$mcV$sp(BatchingExecutor.scala:78) >>> at >>> scala.concurrent.BatchingExecutor$Batch$$anonfun$run$1.apply(BatchingExecutor.scala:55) >>> at >>> scala.concurrent.BatchingExecutor$Batch$$anonfun$run$1.apply(BatchingExecutor.scala:55) >>> at >>> scala.concurrent.BlockContext$.withBlockContext(BlockContext.scala:72) >>> at >>> scala.concurrent.BatchingExecutor$Batch.run(BatchingExecutor.scala:54) >>> at >>> scala.concurrent.Future$InternalCallbackExecutor$.unbatchedExecute(Future.scala:599) >>> at >>> scala.concurrent.BatchingExecutor$class.execute(BatchingExecutor.scala:106) >>> at >>> scala.concurrent.Future$InternalCallbackExecutor$.execute(Future.scala:597) >>> at >>> scala.concurrent.impl.CallbackRunnable.executeWithValue(Promise.scala:40) >>> at >>> scala.concurrent.impl.Promise$DefaultPromise.tryComplete(Promise.scala:248) >>> at scala.concurrent.Promise$class.tryFailure(Promise.scala:112) >>> at >>> scala.concurrent.impl.Promise$DefaultPromise.tryFailure(Promise.scala:153) >>> at >>> org.apache.spark.rpc.netty.NettyRpcEnv$$anon$1.run(NettyRpcEnv.scala:241) >>> at >>> java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) >>> at java.util.concurrent.FutureTask.run(FutureTask.java:266) >>> at >>> java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) >>> at >>> java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) >>> at >>> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) >>> at >>> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) >>> at java.lang.Thread.run(Thread.java:745) >>> Caused by: java.util.concurrent.TimeoutException: Cannot receive any reply >>> in 10 seconds >>> at >>> org.apache.spark.rpc.netty.NettyRpcEnv$$anon$1.run(NettyRpcEnv.scala:242) >>> ... 7 more >>> >>> >>> >>> Here is the config of the spark streaming app >>> >>> val conf = new SparkConf() >>> .setMaster(sparkMaster) >>> .setAppName(sparkApp) >>> .set("spark.cassandra.connection.host", CassandraConfig.host) >>> .set("spark.logConf", "true") >>> .set("spark.fileserver.port","7002") >>> .set("spark.broadcast.port","7003") >>> .set("spark.replClassServer.port","7004") >>> .set("spark.blockManager.port","7005") >>> .set("spark.executor.port","7006") >>> >>> .set("spark.broadcast.factory","org.apache.spark.broadcast.HttpBroadcastFactory") >>> .setJars(sparkJars) >>> >>> val sc = new SparkContext(conf) >>> >>> val ssc = new StreamingContext(sc, Seconds(5)) >>> >>> val tags = ssc.actorStream[String](Props(new >>> GifteeTagStreamingActor("akka.tcp://spark-engine@spark-engine:9083/user/integrationActor")), >>> "TagsReceiver") >>> >>> >>> the docker images for master and worker expose those ports. >>> >>> master ---> EXPOSE 8080 7077 4040 7001 7002 7003 7004 7005 7006 >>> worker ---> EXPOSE 8888 8081 4040 7001 7002 7003 7004 7005 7006 >>> >>> I'm using those images docker images to run spark jobs without a >>> problem. I only get errors on the streaming app. >>> >>> any pointers on what can be wrong? >>> >>> Thank you very much in advanced. >>> >>> David >>> >>> >> >