Fix flaky SharingReferenceTest (#2709)

Previously assert has been racing with the coroutine state spilling, triggering spurious wakeups

Author: qwwdfsad

kotlinx-coroutines-core/jvm/test/flow/SharingReferenceTest.kt

@@ -5,6 +5,7 @@
 package kotlinx.coroutines.flow
 
 import kotlinx.coroutines.*
+import kotlinx.coroutines.internal.*
 import org.junit.*
 
 /**
@@ -15,6 +16,15 @@ import org.junit.*
 class SharingReferenceTest : TestBase() {
     private val token = object {}
 
+    /*
+     * Single-threaded executor that we are using to ensure that the flow being sharing actually
+     * suspended (spilled its locals, attached to parent), so we can verify reachability.
+     * Without that, it's possible to have a situation where target flow is still
+     * being strongly referenced (by its dispatcher), but the test already tries to test reachability and fails.
+     */
+    @get:Rule
+    val executor = ExecutorRule(1)
+
     private val weakEmitter = flow {
         emit(null)
         // suspend forever without keeping a strong reference to continuation -- this is a model of
@@ -26,19 +36,26 @@ class SharingReferenceTest : TestBase() {
 
     @Test
     fun testShareInReference() {
-        val flow = weakEmitter.shareIn(GlobalScope, SharingStarted.Eagerly, 0)
+        val flow = weakEmitter.shareIn(ContextScope(executor), SharingStarted.Eagerly, 0)
+        linearize()
         FieldWalker.assertReachableCount(1, flow) { it === token }
     }
 
     @Test
     fun testStateInReference() {
-        val flow = weakEmitter.stateIn(GlobalScope, SharingStarted.Eagerly, null)
+        val flow = weakEmitter.stateIn(ContextScope(executor), SharingStarted.Eagerly, null)
+        linearize()
         FieldWalker.assertReachableCount(1, flow) { it === token }
     }
 
     @Test
     fun testStateInSuspendingReference() = runTest {
         val flow = weakEmitter.stateIn(GlobalScope)
+        linearize()
         FieldWalker.assertReachableCount(1, flow) { it === token }
     }
-}
+
+    private fun linearize() {
+        runBlocking(executor) {  }
+    }
+}