Add support for quickcheck DStreams

[holdenk]

Add support for quickcheck DStreams.

[mahmoudhanafy]

I tried to implement DStream Generator, but I stuck into a problem. I created a StreamingContext instance and used it to generate InputDStreams. Consider the following code:

val ssc: StreamingContext = new StreamingContext(sc, Seconds(1))
val generator: Gen[InputDStream[Int]] = DStreamGenerator.genInputDStream[Int](ssc)(Arbitrary.arbitrary[Int])

val property =
  forAll(generator) {
    stream => {
      stream.saveAsTextFiles("/tmp/hanafy/")
      ssc.start()
      ssc.awaitTerminationOrTimeout(10000)
      true
    }
  }

Only the first stream is saved and the got the following exception:

Caused by: java.lang.IllegalStateException: Adding new inputs, transformations, and output operations after starting a context is not supported

Only the first InputStream is added before starting the streaming context started, So I tried to close the StreamingContext after saving the stream.

forAll(generator) {
  stream => {
    stream.saveAsTextFiles("/tmp/hanafy/")
    ssc.start()
    ssc.awaitTerminationOrTimeout(10000)
    ssc.stop(stopSparkContext = false)

    true
  }
}

But I got the following exception:

Caused by: java.lang.IllegalStateException: Adding new inputs, transformations, and output operations after stopping a context is not supported

How to deal with this ?

[mahmoudhanafy]

Should I create a new StreamingContext at every stream ?