Moved duplicated code to methods and swapped the order of if else

Author: Efim Poberezkin

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

@@ -142,31 +142,29 @@ private[continuous] class EpochCoordinator(
   private val epochsWaitingToBeCommitted = mutable.HashSet.empty[Long]
 
   private def resolveCommitsAtEpoch(epoch: Long) = {
-    val thisEpochCommits =
-      partitionCommits.collect { case ((e, _), msg) if e == epoch => msg }
+    val thisEpochCommits = findCommitsForEpoch(epoch)
     val nextEpochOffsets =
       partitionOffsets.collect { case ((e, _), o) if e == epoch => o }
 
     if (thisEpochCommits.size == numWriterPartitions &&
       nextEpochOffsets.size == numReaderPartitions) {
 
       // Check that last committed epoch is the previous one for sequencing of committed epochs.
-      if (lastCommittedEpoch == epoch - 1) {
-        logDebug(s"Epoch $epoch has received commits from all partitions. Committing globally.")
-        // Sequencing is important here. We must commit to the writer before recording the commit
-        // in the query, or we will end up dropping the commit if we restart in the middle.
-        writer.commit(epoch, thisEpochCommits.toArray)
-        query.commit(epoch)
+      // If not, add the epoch being currently processed to epochs waiting to be committed,
+      // otherwise commit it.
+      if (lastCommittedEpoch != epoch - 1) {
+        logDebug(s"Epoch $epoch has received commits from all partitions" +
+          s"and is waiting for epoch ${epoch - 1} to be committed first.")
+        epochsWaitingToBeCommitted.add(epoch)
+      } else {
+        commitEpoch(epoch, thisEpochCommits)
         lastCommittedEpoch = epoch
 
         // Commit subsequent epochs that are waiting to be committed.
         var nextEpoch = lastCommittedEpoch + 1
         while (epochsWaitingToBeCommitted.contains(nextEpoch)) {
-          val nextEpochCommits =
-            partitionCommits.collect { case ((e, _), msg) if e == nextEpoch => msg }
-          logDebug(s"Committing epoch $nextEpoch.")
-          writer.commit(nextEpoch, nextEpochCommits.toArray)
-          query.commit(nextEpoch)
+          val nextEpochCommits = findCommitsForEpoch(nextEpoch)
+          commitEpoch(nextEpoch, nextEpochCommits)
 
           epochsWaitingToBeCommitted.remove(nextEpoch)
           lastCommittedEpoch = nextEpoch
@@ -181,14 +179,23 @@ private[continuous] class EpochCoordinator(
         for (k <- partitionOffsets.keys.filter { case (e, _) => e < lastCommittedEpoch }) {
           partitionOffsets.remove(k)
         }
-      } else {
-        logDebug(s"Epoch $epoch has received commits from all partitions" +
-          s"and is waiting for epoch ${epoch - 1} to be committed first.")
-        epochsWaitingToBeCommitted.add(epoch)
       }
     }
   }
 
+  private def findCommitsForEpoch(epoch: Long): Iterable[WriterCommitMessage] = {
+    partitionCommits.collect { case ((e, _), msg) if e == epoch => msg }
+  }
+
+  private def commitEpoch(epoch: Long, messages: Iterable[WriterCommitMessage]): Unit = {
+    logDebug(s"Epoch $epoch has received commits from all partitions" +
+      s"and is ready to be committed. Committing epoch $epoch.")
+    // Sequencing is important here. We must commit to the writer before recording the commit
+    // in the query, or we will end up dropping the commit if we restart in the middle.
+    writer.commit(epoch, messages.toArray)
+    query.commit(epoch)
+  }
+
   override def receive: PartialFunction[Any, Unit] = {
     // If we just drop these messages, we won't do any writes to the query. The lame duck tasks
     // won't shed errors or anything.