{"action":"created","issue":{"url":"https://api.github.com/repos/holdenk/spark-testing-base/issues/27","labels_url":"https://api.github.com/repos/holdenk/spark-testing-base/issues/27/labels{/name}","comments_url":"https://api.github.com/repos/holdenk/spark-testing-base/issues/27/comments","events_url":"https://api.github.com/repos/holdenk/spark-testing-base/issues/27/events","html_url":"https://github.com/holdenk/spark-testing-base/issues/27","id":117443395,"number":27,"title":"Streaming Context is stopped prematurely, which prevents sequence of batches to be executed fully, visible on slower machines","user":{"login":"elyast","id":671809,"avatar_url":"https://avatars.githubusercontent.com/u/671809?v=3","gravatar_id":"","url":"https://api.github.com/users/elyast","html_url":"https://github.com/elyast","followers_url":"https://api.github.com/users/elyast/followers","following_url":"https://api.github.com/users/elyast/following{/other_user}","gists_url":"https://api.github.com/users/elyast/gists{/gist_id}","starred_url":"https://api.github.com/users/elyast/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/elyast/subscriptions","organizations_url":"https://api.github.com/users/elyast/orgs","repos_url":"https://api.github.com/users/elyast/repos","events_url":"https://api.github.com/users/elyast/events{/privacy}","received_events_url":"https://api.github.com/users/elyast/received_events","type":"User","site_admin":false},"labels":[],"state":"closed","locked":false,"assignee":null,"milestone":null,"comments":9,"created_at":"2015-11-17T20:29:41Z","updated_at":"2015-12-30T22:32:24Z","closed_at":"2015-12-30T22:32:24Z","body":"Hi,\r\n\r\nI am struggling with running SparkStreaming tests on slower boxes, and apparently the streaming context is being closed to soon, (before every batch could be processed), my code:\r\n```scala\r\nclass StreamToFileDumperSparkTest  extends StreamingActionBase {\r\n  test(\"smoke integration test for dumper transformation\") {\r\n     val input = List(List(data1), List(data2), List(data3))\r\n     runAction(input, (s: DStream[String]) => s.foreachRDD(/*save some files*/))\r\n     // here I do waiting for files to be available\r\n     val result = sc.textFile(some_path).collect().sorted\r\n     result should equal(inputData)\r\n  }\r\n}\r\n```\r\n\r\nException that shows my suspicion about streaming context be closed to soon is following:\r\n```\r\n15/11/17 12:02:24 INFO DAGScheduler: Submitting ResultStage 1 (MapPartitionsRDD[15] at foreachRDD at StreamToFileDumper.scala:57), which is now runnable\r\nException in thread \"streaming-job-executor-0\" java.lang.Error: java.lang.InterruptedException\r\n\tat java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1151)\r\n\tat java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)\r\n\tat java.lang.Thread.run(Thread.java:745)\r\nCaused by: java.lang.InterruptedException\r\n\tat java.lang.Object.wait(Native Method)\r\n\tat java.lang.Object.wait(Object.java:503)\r\n\tat org.apache.spark.scheduler.JobWaiter.awaitResult(JobWaiter.scala:73)\r\n\tat org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:559)\r\n\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:1822)\r\n\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:1835)\r\n\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:1912)\r\n\tat org.apache.spark.rdd.PairRDDFunctions$$anonfun$saveAsHadoopDataset$1.apply$mcV$sp(PairRDDFunctions.scala:1124)\r\n\tat org.apache.spark.rdd.PairRDDFunctions$$anonfun$saveAsHadoopDataset$1.apply(PairRDDFunctions.scala:1065)\r\n\tat org.apache.spark.rdd.PairRDDFunctions$$anonfun$saveAsHadoopDataset$1.apply(PairRDDFunctions.scala:1065)\r\n\tat org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:147)\r\n\tat org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:108)\r\n\tat org.apache.spark.rdd.RDD.withScope(RDD.scala:306)\r\n\tat org.apache.spark.rdd.PairRDDFunctions.saveAsHadoopDataset(PairRDDFunctions.scala:1065)\r\n\tat org.apache.spark.rdd.PairRDDFunctions$$anonfun$saveAsHadoopFile$4.apply$mcV$sp(PairRDDFunctions.scala:989)\r\n\tat org.apache.spark.rdd.PairRDDFunctions$$anonfun$saveAsHadoopFile$4.apply(PairRDDFunctions.scala:965)\r\n\tat org.apache.spark.rdd.PairRDDFunctions$$anonfun$saveAsHadoopFile$4.apply(PairRDDFunctions.scala:965)\r\n\tat org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:147)\r\n\tat org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:108)\r\n\tat org.apache.spark.rdd.RDD.withScope(RDD.scala:306)\r\n\tat org.apache.spark.rdd.PairRDDFunctions.saveAsHadoopFile(PairRDDFunctions.scala:965)\r\n\tat org.apache.spark.rdd.PairRDDFunctions$$anonfun$saveAsHadoopFile$3.apply$mcV$sp(PairRDDFunctions.scala:951)\r\n\tat org.apache.spark.rdd.PairRDDFunctions$$anonfun$saveAsHadoopFile$3.apply(PairRDDFunctions.scala:951)\r\n\tat org.apache.spark.rdd.PairRDDFunctions$$anonfun$saveAsHadoopFile$3.apply(PairRDDFunctions.scala:951)\r\n\tat org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:147)\r\n\tat org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:108)\r\n\tat org.apache.spark.rdd.RDD.withScope(RDD.scala:306)\r\n\tat org.apache.spark.rdd.PairRDDFunctions.saveAsHadoopFile(PairRDDFunctions.scala:950)\r\n\tat org.apache.spark.rdd.PairRDDFunctions$$anonfun$saveAsHadoopFile$2.apply$mcV$sp(PairRDDFunctions.scala:909)\r\n\tat org.apache.spark.rdd.PairRDDFunctions$$anonfun$saveAsHadoopFile$2.apply(PairRDDFunctions.scala:907)\r\n\tat org.apache.spark.rdd.PairRDDFunctions$$anonfun$saveAsHadoopFile$2.apply(PairRDDFunctions.scala:907)\r\n\tat org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:147)\r\n\tat org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:108)\r\n\tat org.apache.spark.rdd.RDD.withScope(RDD.scala:306)\r\n\tat org.apache.spark.rdd.PairRDDFunctions.saveAsHadoopFile(PairRDDFunctions.scala:907)\r\n\tat org.apache.spark.rdd.RDD$$anonfun$saveAsTextFile$2.apply$mcV$sp(RDD.scala:1444)\r\n\tat org.apache.spark.rdd.RDD$$anonfun$saveAsTextFile$2.apply(RDD.scala:1432)\r\n\tat org.apache.spark.rdd.RDD$$anonfun$saveAsTextFile$2.apply(RDD.scala:1432)\r\n\tat org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:147)\r\n\tat org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:108)\r\n\tat org.apache.spark.rdd.RDD.withScope(RDD.scala:306)\r\n\tat org.apache.spark.rdd.RDD.saveAsTextFile(RDD.scala:1432)\r\n\tat com.nokia.ph.kinesis2s3.spark.StreamToFileDumper.save(StreamToFileDumper.scala:45)\r\n\tat com.nokia.ph.kinesis2s3.spark.StreamToFileDumper$$anonfun$process$1.apply(StreamToFileDumper.scala:57)\r\n\tat com.nokia.ph.kinesis2s3.spark.StreamToFileDumper$$anonfun$process$1.apply(StreamToFileDumper.scala:57)\r\n\tat org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1$$anonfun$apply$mcV$sp$1.apply$mcV$sp(ForEachDStream.scala:42)\r\n\tat org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1$$anonfun$apply$mcV$sp$1.apply(ForEachDStream.scala:40)\r\n\tat org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1$$anonfun$apply$mcV$sp$1.apply(ForEachDStream.scala:40)\r\n\tat org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:399)\r\n\tat org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply$mcV$sp(ForEachDStream.scala:40)\r\n\tat org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply(ForEachDStream.scala:40)\r\n\tat org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply(ForEachDStream.scala:40)\r\n\tat scala.util.Try$.apply(Try.scala:161)\r\n\tat org.apache.spark.streaming.scheduler.Job.run(Job.scala:34)\r\n\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler$$anonfun$run$1.apply$mcV$sp(JobScheduler.scala:218)\r\n\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler$$anonfun$run$1.apply(JobScheduler.scala:218)\r\n\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler$$anonfun$run$1.apply(JobScheduler.scala:218)\r\n\tat scala.util.DynamicVariable.withValue(DynamicVariable.scala:57)\r\n\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.run(JobScheduler.scala:217)\r\n\tat java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)\r\n\t... 2 more\r\n15/11/17 12:02:24 INFO JobScheduler: Stopped JobScheduler\r\n```"},"comment":{"url":"https://api.github.com/repos/holdenk/spark-testing-base/issues/comments/168087257","html_url":"https://github.com/holdenk/spark-testing-base/issues/27#issuecomment-168087257","issue_url":"https://api.github.com/repos/holdenk/spark-testing-base/issues/27","id":168087257,"user":{"login":"elyast","id":671809,"avatar_url":"https://avatars.githubusercontent.com/u/671809?v=3","gravatar_id":"","url":"https://api.github.com/users/elyast","html_url":"https://github.com/elyast","followers_url":"https://api.github.com/users/elyast/followers","following_url":"https://api.github.com/users/elyast/following{/other_user}","gists_url":"https://api.github.com/users/elyast/gists{/gist_id}","starred_url":"https://api.github.com/users/elyast/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/elyast/subscriptions","organizations_url":"https://api.github.com/users/elyast/orgs","repos_url":"https://api.github.com/users/elyast/repos","events_url":"https://api.github.com/users/elyast/events{/privacy}","received_events_url":"https://api.github.com/users/elyast/received_events","type":"User","site_admin":false},"created_at":"2015-12-30T22:32:24Z","updated_at":"2015-12-30T22:32:24Z","body":"its been merged already closing"}}