util: Run QP.wrappingProcessor() continuation inside finally block

platform/platform-api/src/com/intellij/util/concurrency/QueueProcessor.java

@@ -73,9 +73,13 @@ public class QueueProcessor<T> {
 
   @NotNull
   private static <T> PairConsumer<T, Runnable> wrappingProcessor(@NotNull final Consumer<T> processor) {
-    return (item, runnable) -> {
-      runSafely(() -> processor.consume(item));
-      runnable.run();
+    return (item, continuation) -> {
+      try {
+        runSafely(() -> processor.consume(item));
+      }
+      finally {
+        continuation.run();
+      }
     };
   }
 

platform/platform-tests/testSrc/com/intellij/util/concurrency/QueueProcessorTest.kt

@@ -15,9 +15,17 @@
  */
 package com.intellij.util.concurrency
 
+import com.intellij.execution.ExecutionException
+import com.intellij.openapi.progress.ProcessCanceledException
+import com.intellij.testFramework.LoggedErrorProcessor
 import com.intellij.testFramework.PlatformTestCase
+import java.util.concurrent.LinkedBlockingQueue
+import java.util.concurrent.TimeUnit
+
+private const val TIMEOUT_MS = 1000L
 
 class QueueProcessorTest : PlatformTestCase() {
+
   fun `test waiting for returns on finish condition`() {
     var stop = false;
     val semaphore = Semaphore(0)
@@ -26,8 +34,49 @@ class QueueProcessorTest : PlatformTestCase() {
     processor.add(1)
     stop = true;
     semaphore.up()
-    
-    assertTrue(processor.waitFor(1000));
+
+    assertTrue(processor.waitFor(TIMEOUT_MS));
     processor.waitFor(); // just in case let's check this method as well - hopefully, it won't hang since waitFor(timeout) works
   }
+
+  fun `test works fine after thrown exception`() {
+    LoggedErrorProcessor.getInstance().disableStderrDumping(testRootDisposable)
+
+    val resultQueue = LinkedBlockingQueue<Any>()
+    val queueProcessor = QueueProcessor<() -> Any> {
+      try {
+        resultQueue.add(it())
+      }
+      catch (e: Throwable) {
+        resultQueue.add(e)
+        throw e
+      }
+    }
+
+    fun check(expectedResult: Any, item: () -> Any) {
+      queueProcessor.add(item)
+      assertEquals(expectedResult, resultQueue.poll(TIMEOUT_MS, TimeUnit.MILLISECONDS))
+      assertEmpty(resultQueue)
+    }
+    fun check(expectedResult: Number) = check(expectedResult) { expectedResult }
+    fun check(expectedException: Throwable) = check(expectedException) {
+      throw expectedException.also {
+        it.addSuppressed(Throwable())
+      }
+    }
+
+    check(1)
+    check(Throwable())
+    check(2)
+    check(Error())
+    check(3)
+    check(RuntimeException())
+    check(4)
+    check(ProcessCanceledException())  // this used to make the remaining queue elements stuck
+    check(5)
+    check(Exception())
+    check(6)
+    check(ExecutionException("EE"))
+    check(7)
+  }
 }