I would have expected this test to fail:

 @Test
fun testFlowFailure() = runTest {
    val flow = MutableStateFlow(0)
    val latch = CountDownLatch(1)
    val job = launch {
        flow.collect {
            if (it == 1) {
                fail()
                latch.countDown()
            }
        }
    }
    flow.emit(1)
    latch.await()
    job.cancelAndJoin()
}

I do get the expected failure if I do countDown before fail(), so I assume that the exception is postponed until the join call. How can I change the code so it fails immediately inside the collect?