title: Spark streaming store method only work in Duration window but not in foreachRDD workflow in customized receiver
body: <p>I define a receiver to read data from Redis.</p>
  
  <p>part of receiver simplified code:</p>
  
  <pre><code>class MyReceiver extends Receiver (StorageLevel.MEMORY_ONLY){
    override def onStart() = {
      while(!isStopped) {
        val res = readMethod()
        if (res != null) store(res.toIterator) 
        // using res.foreach(r =&gt; store(r)) the performance is almost the same
      }
    }
  }
  </code></pre>
  
  <p>My streaming workflow:</p>
  
  <pre><code>val ssc = new StreamingContext(spark.sparkContext, new Duration(50))
  val myReceiver = new MyReceiver()
  val s = ssc.receiverStream(myReceiver)
  s.foreachRDD{ r =&gt; 
    r.persist()
    if (!r.isEmpty) {
      some short operations about 1s in total
      // note this line ######1
    }
  }
  </code></pre>
  
  <p>I have a producer which produce much faster than consumer so that there are plenty records in Redis now, I tested with number 10000. I debugged, and all records could quickly be read after they are in Redis by <code>readMethod()</code> above. However, <strong>in each microbatch I can only get 30 records</strong>. (If store is fast enough it should get all of 10000)</p>
  
  <p>With this suspect, I added a sleep 10 seconds code <code>Thread.sleep(10000)</code> to <code>######1</code> above. <strong>Each microbatch still gets about 30 records</strong>, and each microbatch process time increases 10 seconds. And if I increase the Duration to 200ms, <code>val ssc = new StreamingContext(spark.sparkContext, new Duration(200))</code>, it could get about 120 records.</p>
  
  <p>All of these shows spark streaming only generate RDD in <code>Duration</code>? After gets RDD and in main workflow, <code>store</code> method is temporarily stopped? But this is a great waste if it is true. I want it also generates RDD (store) while the main workflow is running. </p>
  
  <p>Any ideas?</p>