show missed input info in streaming info page

Author: wei-mao-intel

streaming/src/main/scala/org/apache/spark/streaming/scheduler/JobScheduler.scala

@@ -21,6 +21,7 @@ import java.util.Properties
 import java.util.concurrent.{ConcurrentHashMap, TimeUnit}
 
 import scala.collection.JavaConverters._
+import scala.collection.mutable.HashSet
 import scala.util.Failure
 
 import org.apache.commons.lang.SerializationUtils
@@ -64,6 +65,8 @@ class JobScheduler(val ssc: StreamingContext) extends Logging {
 
   private var eventLoop: EventLoop[JobSchedulerEvent] = null
 
+  private val inputInfoMissedTimes = HashSet[Time]()
+
   def start(): Unit = synchronized {
     if (eventLoop != null) return // scheduler has already been started
 
@@ -139,6 +142,7 @@ class JobScheduler(val ssc: StreamingContext) extends Logging {
   def submitJobSet(jobSet: JobSet) {
     if (jobSet.jobs.isEmpty) {
       logInfo("No jobs added for time " + jobSet.time)
+      inputInfoMissedTimes.add(jobSet.time)
     } else {
       listenerBus.post(StreamingListenerBatchSubmitted(jobSet.toBatchInfo))
       jobSets.put(jobSet.time, jobSet)
@@ -193,6 +197,14 @@ class JobScheduler(val ssc: StreamingContext) extends Logging {
     listenerBus.post(StreamingListenerOutputOperationCompleted(job.toOutputOperationInfo))
     logInfo("Finished job " + job.id + " from job set of time " + jobSet.time)
     if (jobSet.hasCompleted) {
+      // submit fake BatchCompleted event to show missing inputInfo on Streaming UI
+      inputInfoMissedTimes.foreach (time => {
+        val streamIdToInputInfos = inputInfoTracker.getInfo(time)
+        val fakeJobSet = JobSet(time, Seq(), streamIdToInputInfos)
+        listenerBus.post(StreamingListenerBatchCompleted(fakeJobSet.toBatchInfo))
+      })
+      inputInfoMissedTimes.clear()
+
       jobSets.remove(jobSet.time)
       jobGenerator.onBatchCompletion(jobSet.time)
       logInfo("Total delay: %.3f s for time %s (execution: %.3f s)".format(

streaming/src/main/scala/org/apache/spark/streaming/scheduler/JobSet.scala

@@ -50,7 +50,7 @@ case class JobSet(
 
   def hasStarted: Boolean = processingStartTime > 0
 
-  def hasCompleted: Boolean = incompleteJobs.isEmpty
+  def hasCompleted: Boolean = incompleteJobs.isEmpty && processingStartTime >= 0
 
   // Time taken to process all the jobs from the time they started processing
   // (i.e. not including the time they wait in the streaming scheduler queue)