Fixed bug

Author: tdas

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

@@ -543,6 +543,15 @@ abstract class StreamExecution(
     Option(name).map(_ + "<br/>").getOrElse("") +
       s"id = $id<br/>runId = $runId<br/>batch = $batchDescription"
   }
+
+  private[sql] def withProgressLocked(f: => Unit): Unit = {
+    awaitProgressLock.lock()
+    try {
+      f
+    } finally {
+      awaitProgressLock.unlock()
+    }
+  }
 }
 
 object StreamExecution {

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

@@ -282,19 +282,18 @@ class ContinuousExecution(
       epoch: Long, reader: ContinuousReader, partitionOffsets: Seq[PartitionOffset]): Unit = {
     assert(continuousSources.length == 1, "only one continuous source supported currently")
 
-    val globalOffset = reader.mergeOffsets(partitionOffsets.toArray)
-    val oldOffset = synchronized {
-      offsetLog.add(epoch, OffsetSeq.fill(globalOffset))
-      offsetLog.get(epoch - 1)
-    }
-
-    // If offset hasn't changed since last epoch, there's been no new data.
-    if (oldOffset.contains(OffsetSeq.fill(globalOffset))) {
-      noNewData = true
-    }
-
     awaitProgressLock.lock()
     try {
+      val globalOffset = reader.mergeOffsets(partitionOffsets.toArray)
+      val oldOffset = synchronized {
+        offsetLog.add(epoch, OffsetSeq.fill(globalOffset))
+        offsetLog.get(epoch - 1)
+      }
+
+      // If offset hasn't changed since last epoch, there's been no new data.
+      if (oldOffset.contains(OffsetSeq.fill(globalOffset))) {
+        noNewData = true
+      }
       awaitProgressLockCondition.signalAll()
     } finally {
       awaitProgressLock.unlock()
@@ -308,23 +307,22 @@ class ContinuousExecution(
   def commit(epoch: Long): Unit = {
     assert(continuousSources.length == 1, "only one continuous source supported currently")
     assert(offsetLog.get(epoch).isDefined, s"offset for epoch $epoch not reported before commit")
-    synchronized {
+
+    awaitProgressLock.lock()
+    try {
       if (queryExecutionThread.isAlive) {
         commitLog.add(epoch)
         val offset = offsetLog.get(epoch).get.offsets(0).get
         committedOffsets ++= Seq(continuousSources(0) -> offset)
       } else {
         return
       }
-    }
 
-    if (minLogEntriesToMaintain < currentBatchId) {
-      offsetLog.purge(currentBatchId - minLogEntriesToMaintain)
-      commitLog.purge(currentBatchId - minLogEntriesToMaintain)
-    }
+      if (minLogEntriesToMaintain < currentBatchId) {
+        offsetLog.purge(currentBatchId - minLogEntriesToMaintain)
+        commitLog.purge(currentBatchId - minLogEntriesToMaintain)
+      }
 
-    awaitProgressLock.lock()
-    try {
       awaitProgressLockCondition.signalAll()
     } finally {
       awaitProgressLock.unlock()