Address TD's comments

Author: zsxwing

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamExecution.scala

@@ -63,7 +63,7 @@ class StreamExecution(
 
   private val pollingDelayMs = sparkSession.sessionState.conf.streamingPollingDelay
 
-  private val noDataReportInterval = sparkSession.sessionState.conf.streamingNoDataReportInterval
+  private val noDataEventInterval = sparkSession.sessionState.conf.streamingNoDataEventInterval
 
   /**
    * A lock used to wait/notify when batches complete. Use a fair lock to avoid thread starvation.
@@ -199,7 +199,7 @@ class StreamExecution(
       SparkSession.setActiveSession(sparkSession)
 
       // The timestamp we report an event that has no input data
-      var noDataReportTimestamp = Long.MinValue
+      var noDataEventTimestamp = Long.MinValue
 
       triggerExecutor.execute(() => {
         startTrigger()
@@ -224,11 +224,13 @@ class StreamExecution(
             // Report trigger as finished and construct progress object.
             finishTrigger(dataAvailable)
             if (dataAvailable) {
+              // Reset noDataEventTimestamp if we processed any data
+              noDataEventTimestamp = Long.MinValue
               postEvent(new QueryProgressEvent(lastProgress))
             } else {
               val now = triggerClock.getTimeMillis()
-              if (now - noDataReportInterval >= noDataReportTimestamp) {
-                noDataReportTimestamp = now
+              if (now - noDataEventInterval >= noDataEventTimestamp) {
+                noDataEventTimestamp = now
                 postEvent(new QueryProgressEvent(lastProgress))
               }
             }

sql/core/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -603,8 +603,8 @@ object SQLConf {
       .timeConf(TimeUnit.MILLISECONDS)
       .createWithDefault(10L)
 
-  val STREAMING_NO_DATA_REPORT_INTERVAL =
-    SQLConfigBuilder("spark.sql.streaming.noDataReportInterval")
+  val STREAMING_NO_DATA_EVENT_INTERVAL =
+    SQLConfigBuilder("spark.sql.streaming.noDataEventInterval")
       .internal()
       .doc("How long to wait between two progress events when there is no data")
       .timeConf(TimeUnit.MILLISECONDS)
@@ -691,7 +691,7 @@ private[sql] class SQLConf extends Serializable with CatalystConf with Logging {
 
   def streamingPollingDelay: Long = getConf(STREAMING_POLLING_DELAY)
 
-  def streamingNoDataReportInterval: Long = getConf(STREAMING_NO_DATA_REPORT_INTERVAL)
+  def streamingNoDataEventInterval: Long = getConf(STREAMING_NO_DATA_EVENT_INTERVAL)
 
   def streamingMetricsEnabled: Boolean = getConf(STREAMING_METRICS_ENABLED)
 

sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingQueryListenerSuite.scala

@@ -192,8 +192,9 @@ class StreamingQueryListenerSuite extends StreamTest with BeforeAndAfter {
     assert(queryQueryTerminated.exception === newQueryTerminated.exception)
   }
 
-  test("noDataReportInterval") {
-    withSQLConf(SQLConf.STREAMING_NO_DATA_REPORT_INTERVAL.key -> "100ms") {
+  test("only one progress event per interval when no data") {
+    // This test will start a query but not push any data, and then check if we push too many events
+    withSQLConf(SQLConf.STREAMING_NO_DATA_EVENT_INTERVAL.key -> "100ms") {
       @volatile var progressEventCount = 0
 
       val listener = new StreamingQueryListener {
@@ -220,13 +221,14 @@ class StreamingQueryListenerSuite extends StreamTest with BeforeAndAfter {
         for (_ <- 1 to 100) {
           actions += AdvanceManualClock(10)
           actions += AssertOnQuery { _ =>
-            // Sleep so that if the config `noDataReportInterval` doesn't work, it has enough time
+            // Sleep so that if the config `noDataEventInterval` doesn't work, it has enough time
             // to report too many events.
             Thread.sleep(10)
             true
           }
         }
         testStream(MemoryStream[Int].toDS)(actions: _*)
+        // 11 is the max value of the possible numbers of events.
         assert(progressEventCount <= 11)
       } finally {
         spark.streams.removeListener(listener)