ronantakizawa/codereview-bench huggingface dataset:
Repository: Kotlin/kotlinx.coroutines
File: kotlinx-coroutines-core/common/src/intrinsics/Undispatched.kt (Kotlin)

Code before:
     * We are trying to complete our undispatched block with the following possible codepaths:
     * 1) The coroutine just suspended. I.e. `coroutineScope { .. suspend here }`.
     *   Then just suspend
     * 2) The coroutine completed with something, but has active children. Wait for them, also suspend
     * 3) The coroutine succesfully completed. Return or rethrow its result.
     */
    if (result === COROUTINE_SUSPENDED) return COROUTINE_SUSPENDED // (1)
    val state = makeCompletingOnce(result)
    if (state === COMPLETING_WAITING_CHILDREN) return COROUTINE_SUSPENDED // (2)
    return if (state is CompletedExceptionally) { // (3)
        when {
            alwaysRethrow || notOwnTimeout(state.cause) -> throw recoverStackTrace(state.cause, uCont)
            result is CompletedExceptionally -> throw recoverStackTrace(result.cause, uCont)
            else -> result
        }
    } else {
        state.unboxState()
    }
}

private fun ScopeCoroutine<*>.notOwnTimeout(cause: Throwable): Boolean {
    return cause !is TimeoutCancellationException || cause.coroutine !== this
}

private fun ScopeCoroutine<*>.dispatchException(e: DispatchException) {
    makeCompleting(CompletedExceptionally(e.cause))
    throw throw recoverStackTrace(e.cause, uCont)
}


Review comment (suggestion):
I think this function should be inlined. It's not obvious that it calls `makeCompleting`.

Code after:
     * We are trying to complete our undispatched block with the following possible codepaths:
     * 1) The coroutine just suspended. I.e. `coroutineScope { .. suspend here }`.
     *   Then just suspend
     * 2) The coroutine completed with something, but has active children. Wait for them, also suspend
     * 3) The coroutine succesfully completed. Return or rethrow its result.
     */
    if (result === COROUTINE_SUSPENDED) return COROUTINE_SUSPENDED // (1)
    val state = makeCompletingOnce(result)
    if (state === COMPLETING_WAITING_CHILDREN) return COROUTINE_SUSPENDED // (2)
    afterCompletionUndispatched()
    return if (state is CompletedExceptionally) { // (3)
        when {
            alwaysRethrow || notOwnTimeout(state.cause) -> throw recoverStackTrace(state.cause, uCont)
            result is CompletedExceptionally -> throw recoverStackTrace(result.cause, uCont)
            else -> result
        }
    } else {
        state.unboxState()
    }
}

private fun ScopeCoroutine<*>.notOwnTimeout(cause: Throwable): Boolean {
    return cause !is TimeoutCancellationException || cause.coroutine !== this
}

private fun ScopeCoroutine<*>.dispatchExceptionAndMakeCompleting(e: DispatchException): Nothing {
    makeCompleting(CompletedExceptionally(e.cause))
    throw recoverStackTrace(e.cause, uCont)
}
Repository: Kotlin/kotlinx.coroutines
File: kotlinx-coroutines-core/jvm/src/flow/internal/FlowExceptions.kt (Kotlin)

Code before:
package kotlinx.coroutines.flow.internal

import kotlinx.coroutines.*

internal actual class AbortFlowException actual constructor(
    @JvmField @Transient actual val owner: Any
) : CancellationException("Flow was aborted, no more elements needed") {

    override fun fillInStackTrace(): Throwable {
        if (DEBUG) return super.fillInStackTrace()
        // Prevent Android <= 6.0 bug, #1866
        stackTrace = emptyArray()
        return this
    }
}

internal actual class ChildCancelledException : CancellationException("Child of the scoped flow was cancelled") {
    override fun fillInStackTrace(): Throwable {
        if (DEBUG) return super.fillInStackTrace()
        // Prevent Android <= 6.0 bug, #1866
        stackTrace = emptyArray()
        return this
    }
}


Review comment (bug):
It seems possible to maliciously deserialize and use an `AbortFlowException` that would fail in ways that legal instances wouldn't fail. I'm not sure why not just mark `owner` as nullable: I don't think anything breaks if we have `@JvmField @Transient val owner: Any? = owner` where a constructor only accepts a non-nullable version. Why not do that instead?

Code after:
package kotlinx.coroutines.flow.internal

import kotlinx.coroutines.*

/**
 * Implementation note: `owner` is an internal marked that is used ONLY for identity checks by coroutines machinery,
 * and it's never exposed, thus it's safe to have it both `@Transient` and non-nullable.
 */
internal actual class AbortFlowException actual constructor(
    @JvmField @Transient actual val owner: Any
) : CancellationException("Flow was aborted, no more elements needed") {

    override fun fillInStackTrace(): Throwable {
        if (DEBUG) return super.fillInStackTrace()
        // Prevent Android <= 6.0 bug, #1866
        stackTrace = emptyArray()
        return this
    }
}

internal actual class ChildCancelledException : CancellationException("Child of the scoped flow was cancelled") {
    override fun fillInStackTrace(): Throwable {
        if (DEBUG) return super.fillInStackTrace()
        // Prevent Android <= 6.0 bug, #1866
        stackTrace = emptyArray()
        return this
    }
}
Repository: Kotlin/kotlinx.coroutines
File: kotlinx-coroutines-core/common/test/flow/sharing/StateInTest.kt (Kotlin)

Code before:
        assertFailsWith<TestException> { flow.stateIn(CoroutineScope(currentCoroutineContext() + Job() + ceh)) }
        finish(3)
    }

    @Test
    fun testSubscriptionByFirstSuspensionInStateFlow() = runTest {
        testSubscriptionByFirstSuspensionInCollect(flowOf(1).stateIn(this@runTest)) { }
    }

    @Test
    fun testRethrowsCEOnCancelledScope() = runTest {
        val cancelledScope = CoroutineScope(EmptyCoroutineContext).apply { cancel("CancelMessageToken") }
        val flow = flowOf(1, 2, 3)
        assertFailsWith<CancellationException>("CancelMessageToken") {
            flow.stateIn(cancelledScope)
        }
    }

    @Test
    fun testThrowsNoSuchElementExceptionOnEmptyFlow() = runTest {
        val flow = emptyFlow<Any>()
        assertFailsWith<NoSuchElementException> {
            flow.stateIn(this)
        }
        // Ensure that the collecting scope is not cancelled by the NoSuchElementException
        assertTrue(coroutineContext[Job]?.isActive)
    }
}


Review comment (suggestion):
```suggestion
        assertEquals(true, coroutineContext[Job]?.isActive)
```

Code after:
        assertFailsWith<TestException> { flow.stateIn(CoroutineScope(currentCoroutineContext() + Job() + ceh)) }
        finish(3)
    }

    @Test
    fun testSubscriptionByFirstSuspensionInStateFlow() = runTest {
        testSubscriptionByFirstSuspensionInCollect(flowOf(1).stateIn(this@runTest)) { }
    }

    @Test
    fun testRethrowsCEOnCancelledScope() = runTest {
        val cancelledScope = CoroutineScope(EmptyCoroutineContext).apply { cancel("CancelMessageToken") }
        val flow = flowOf(1, 2, 3)
        assertFailsWith<CancellationException>("CancelMessageToken") {
            flow.stateIn(cancelledScope)
        }
    }

    @Test
    fun testThrowsNoSuchElementExceptionOnEmptyFlow() = runTest {
        val flow = emptyFlow<Any>()
        assertFailsWith<NoSuchElementException> {
            flow.stateIn(this)
        }
        // Ensure that the collecting scope is not cancelled by the NoSuchElementException
        assertEquals(true, coroutineContext[Job]?.isActive)
    }
}
Repository: Kotlin/kotlinx.coroutines
File: integration-testing/src/javaConsumersTest/java/RunBlockingJavaTest.java (Java)

Code before:
import kotlinx.coroutines.BuildersKt;
import kotlinx.coroutines.Dispatchers;
import org.junit.Test;

public class RunBlockingJavaTest {
    Boolean entered = false;

    /** This code will not compile if `runBlocking` doesn't declare `@Throws(InterruptedException::class)` */
    @Test
    public void testRunBlocking() {
        try {
            BuildersKt.runBlocking(Dispatchers.getIO(), (scope, continuation) -> {
                entered = true;
                return null;
            });
        } catch (InterruptedException e) {
        }
        assert entered;
    }
}


Review comment (suggestion):
Consider using JUnit's `assertFoo` functions. Java's `assert`s can be disabled (or, to be more precise, can be not enabled). Especially for our custom Gradle setup, the simpler the thing -- the better

Code after:
import kotlinx.coroutines.BuildersKt;
import kotlinx.coroutines.Dispatchers;
import org.junit.Test;
import org.junit.Assert;

public class RunBlockingJavaTest {
    Boolean entered = false;

    /** This code will not compile if `runBlocking` doesn't declare `@Throws(InterruptedException::class)` */
    @Test
    public void testRunBlocking() {
        try {
            BuildersKt.runBlocking(Dispatchers.getIO(), (scope, continuation) -> {
                entered = true;
                return null;
            });
        } catch (InterruptedException e) {
        }
        Assert.assertTrue(entered);
    }
}
Repository: Kotlin/kotlinx.coroutines
File: kotlinx-coroutines-core/common/src/flow/operators/Lint.kt (Kotlin)

Code before:
/**
 * A specialized version of [Flow.toSet] that returns [Nothing]
 * to indicate that [SharedFlow] collection never completes.
 */
@InlineOnly
public suspend inline fun <T> SharedFlow<T>.toSet(destination: MutableSet<T>): Nothing {
    (this as Flow<T>).toSet(destination)
    throw IllegalStateException("this code is supposed to be unreachable")
}

/**
 * @suppress
 */
@Suppress("DeprecatedCallableAddReplaceWith")
@Deprecated(
    message = "SharedFlow never completes, so this terminal operation never completes.",
    level = DeprecationLevel.WARNING
)
@InlineOnly
public suspend inline fun <T> SharedFlow<T>.count(): Int =
    (this as Flow<T>).count()

/**
 * @suppress
 */
@Suppress("DeprecatedCallableAddReplaceWith")
@Deprecated(
    level = DeprecationLevel.WARNING,
    message = "SharedFlow never completes, so this terminal operation never completes. Use 'replayCache.last()' for reading the last emitted item in the replay cache.",
    replaceWith = ReplaceWith("replayCache.last()")
)
@InlineOnly
public fun <T> SharedFlow<T>.last(): T = noImpl()


Review comment (suggestion):
Good call making this a `WARNING` and not an `ERROR`. Wouldn't be nice to break the compilation of someone's code using `last()` instead of `collect()`.

Code after:
 * to indicate that [SharedFlow] collection never completes.
 */
@InlineOnly
public suspend inline fun <T> SharedFlow<T>.toSet(destination: MutableSet<T>): Nothing {
    (this as Flow<T>).toSet(destination)
    throw IllegalStateException("this code is supposed to be unreachable")
}

/**
 * @suppress
 */
@Deprecated(
    message = "SharedFlow never completes, so this terminal operation never completes.",
    level = DeprecationLevel.WARNING
)
@InlineOnly
public suspend inline fun <T> SharedFlow<T>.count(): Int =
    (this as Flow<T>).count()

/**
 * @suppress
 */
@Deprecated(
    message = "SharedFlow never completes, so this terminal operation never completes. " +
        "If you are using last() to imitate a subscriber, use collect() instead.",
    level = DeprecationLevel.WARNING,
)
@InlineOnly
public suspend inline fun <T> SharedFlow<T>.last(): T =
    (this as Flow<T>).last()
Repository: Kotlin/kotlinx.coroutines
File: kotlinx-coroutines-core/common/src/flow/operators/Lint.kt (Kotlin)

Code before:
@InlineOnly
public suspend inline fun <T> SharedFlow<T>.toSet(destination: MutableSet<T>): Nothing {
    (this as Flow<T>).toSet(destination)
    throw IllegalStateException("this code is supposed to be unreachable")
}

/**
 * @suppress
 */
@Suppress("DeprecatedCallableAddReplaceWith")
@Deprecated(
    message = "SharedFlow never completes, so this terminal operation never completes.",
    level = DeprecationLevel.WARNING
)
@InlineOnly
public suspend inline fun <T> SharedFlow<T>.count(): Int =
    (this as Flow<T>).count()

/**
 * @suppress
 */
@Suppress("DeprecatedCallableAddReplaceWith")
@Deprecated(
    level = DeprecationLevel.WARNING,
    message = "SharedFlow never completes, so this terminal operation never completes. Use 'replayCache.last()' for reading the last emitted item in the replay cache.",
    replaceWith = ReplaceWith("replayCache.last()")
)
@InlineOnly
public fun <T> SharedFlow<T>.last(): T = noImpl()


Review comment (performance):
`ReplaceWith` should typically be used for suggesting code that's equivalent to the deprecated one, but more explicitly written to highlight the problem. Example: https://kotlinlang.org/api/core/kotlin-stdlib/kotlin.text/capitalize.html
The reason is that people can perform replacements without looking closely, and the changed behavior can be a surprise. The proper replacement would be something like `(this as Flow<T>).last()`, but I'm not sure it's worth providing, as it is meaningless anyway.

Another problem with this replacement specifically is that `replayCache` is often empty (for example, https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/-mutable-shared-flow.html has its size `0` by default), and `last()` throws exceptions on empty lists.

Code after:
    (this as Flow<T>).toSet(destination)
    throw IllegalStateException("this code is supposed to be unreachable")
}

/**
 * @suppress
 */
@Deprecated(
    message = "SharedFlow never completes, so this terminal operation never completes.",
    level = DeprecationLevel.WARNING
)
@InlineOnly
public suspend inline fun <T> SharedFlow<T>.count(): Int =
    (this as Flow<T>).count()

/**
 * @suppress
 */
@Deprecated(
    message = "SharedFlow never completes, so this terminal operation never completes. " +
        "If you are using last() to imitate a subscriber, use collect() instead.",
    level = DeprecationLevel.WARNING,
)
@InlineOnly
public suspend inline fun <T> SharedFlow<T>.last(): T =
    (this as Flow<T>).last()
Repository: Kotlin/kotlinx.coroutines
File: kotlinx-coroutines-core/common/src/flow/operators/Lint.kt (Kotlin)

Code before:
 */
@InlineOnly
public suspend inline fun <T> SharedFlow<T>.toSet(destination: MutableSet<T>): Nothing {
    (this as Flow<T>).toSet(destination)
    throw IllegalStateException("this code is supposed to be unreachable")
}

/**
 * @suppress
 */
@Suppress("DeprecatedCallableAddReplaceWith")
@Deprecated(
    message = "SharedFlow never completes, so this terminal operation never completes.",
    level = DeprecationLevel.WARNING
)
@InlineOnly
public suspend inline fun <T> SharedFlow<T>.count(): Int =
    (this as Flow<T>).count()

/**
 * @suppress
 */
@Suppress("DeprecatedCallableAddReplaceWith")
@Deprecated(
    level = DeprecationLevel.WARNING,
    message = "SharedFlow never completes, so this terminal operation never completes. Use 'replayCache.last()' for reading the last emitted item in the replay cache.",
    replaceWith = ReplaceWith("replayCache.last()")
)
@InlineOnly
public fun <T> SharedFlow<T>.last(): T = noImpl()


Review comment (suggestion):
One use case for `sharedFlow.last()` that comes to mind is to subscribe to a shared flow so that it "starts working" (as shared flows can be configured behave differently depending on whether there are subscribers: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/-sharing-started/), but doesn't actually do anything. In other words, subscribing for the side effects of subscribing and not for the value. In that scenario, the idiomatic approach is to call `sharedFlow.collect()`. This could be worth highlighting in the replacement message.

Code after:
public suspend inline fun <T> SharedFlow<T>.toSet(destination: MutableSet<T>): Nothing {
    (this as Flow<T>).toSet(destination)
    throw IllegalStateException("this code is supposed to be unreachable")
}

/**
 * @suppress
 */
@Deprecated(
    message = "SharedFlow never completes, so this terminal operation never completes.",
    level = DeprecationLevel.WARNING
)
@InlineOnly
public suspend inline fun <T> SharedFlow<T>.count(): Int =
    (this as Flow<T>).count()

/**
 * @suppress
 */
@Deprecated(
    message = "SharedFlow never completes, so this terminal operation never completes. " +
        "If you are using last() to imitate a subscriber, use collect() instead.",
    level = DeprecationLevel.WARNING,
)
@InlineOnly
public suspend inline fun <T> SharedFlow<T>.last(): T =
    (this as Flow<T>).last()
Repository: Kotlin/kotlinx.coroutines
File: kotlinx-coroutines-core/common/src/flow/operators/Lint.kt (Kotlin)

Code before:
/**
 * A specialized version of [Flow.toSet] that returns [Nothing]
 * to indicate that [SharedFlow] collection never completes.
 */
@InlineOnly
public suspend inline fun <T> SharedFlow<T>.toSet(destination: MutableSet<T>): Nothing {
    (this as Flow<T>).toSet(destination)
    throw IllegalStateException("this code is supposed to be unreachable")
}

/**
 * @suppress
 */
@Suppress("DeprecatedCallableAddReplaceWith")
@Deprecated(
    message = "SharedFlow never completes, so this terminal operation never completes.",
    level = DeprecationLevel.WARNING
)
@InlineOnly
public suspend inline fun <T> SharedFlow<T>.count(): Int =
    (this as Flow<T>).count()

/**
 * @suppress
 */
@Suppress("DeprecatedCallableAddReplaceWith")
@Deprecated(
    message = "SharedFlow never completes, so this terminal operation never completes. " +
        "If you are using last() to imitate a subscriber, use collect() instead.",
    level = DeprecationLevel.WARNING,
)
@InlineOnly
public suspend inline fun <T> SharedFlow<T>.last(): T =
    (this as Flow<T>).last()


Review comment (bug):
The purpose of this annotation is that in `kotlinx.coroutines`, all warnings in the non-test source sets lead to a build failure, so we have to suppress false warnings and can not just ignore them.
That said, the warning suppressed by this no longer exists, so this line does nothing.
Could you please remove this suppression *everywhere in the codebase*?

Code after:
 * to indicate that [SharedFlow] collection never completes.
 */
@InlineOnly
public suspend inline fun <T> SharedFlow<T>.toSet(destination: MutableSet<T>): Nothing {
    (this as Flow<T>).toSet(destination)
    throw IllegalStateException("this code is supposed to be unreachable")
}

/**
 * @suppress
 */
@Deprecated(
    message = "SharedFlow never completes, so this terminal operation never completes.",
    level = DeprecationLevel.WARNING
)
@InlineOnly
public suspend inline fun <T> SharedFlow<T>.count(): Int =
    (this as Flow<T>).count()

/**
 * @suppress
 */
@Deprecated(
    message = "SharedFlow never completes, so this terminal operation never completes. " +
        "If you are using last() to imitate a subscriber, use collect() instead.",
    level = DeprecationLevel.WARNING,
)
@InlineOnly
public suspend inline fun <T> SharedFlow<T>.last(): T =
    (this as Flow<T>).last()
Repository: Kotlin/kotlinx.coroutines
File: kotlinx-coroutines-core/js/test/PropagateExceptionFinalResortTest.kt (Kotlin)

Code before:
import kotlin.test.*

class PropagateExceptionFinalResortTest : TestBase() {
    @BeforeTest
    fun setupExceptionListenerOnly() {
        // Remove a Node.js's internal listener, which prints the exception to stdout.
        js("""
            globalThis.originalListeners = process.listeners('uncaughtException');
            process.removeAllListeners('uncaughtException');
        """)
    }

    @AfterTest
    fun restoreListeners() {
        js("""
            if (globalThis.originalListeners) {
                process.removeAllListeners('uncaughtException');
                globalThis.originalListeners.forEach(function(listener) {
                    process.on('uncaughtException', listener);
                });
            }
        """)
    }

    @Test
    fun testThrows() = runTest {
        js("""
            globalThis.exceptionCaught = false;
            process.on('uncaughtException', function(e) {
                globalThis.exceptionCaught = true;
            });
        """)
        val job = GlobalScope.launch {
            throw IllegalStateException("My ISE")
        }
        job.join()
        delay(1)
        val exceptionCaught = js("globalThis.exceptionCaught") as Boolean
        assertTrue(exceptionCaught)
    }
}

Review comment (suggestion):
I'd expect a more descriptive name, like `testHandlingUncaughtExceptionsInNodeJs`. Maybe even a documentation comment explaining what exactly we are checking here. Given how much JS code there is in this file, it's not trivial to figure out.

Code after:
import kotlin.test.*

class PropagateExceptionFinalResortTest : TestBase() {
    @BeforeTest
    private fun removeListeners() {
        // Remove a Node.js's internal listener, which prints the exception to stdout.
        js("""
            globalThis.originalListeners = process.listeners('uncaughtException');
            process.removeAllListeners('uncaughtException');
        """)
    }

    @AfterTest
    private fun restoreListeners() {
        js("""
            if (globalThis.originalListeners) {
                process.removeAllListeners('uncaughtException');
                globalThis.originalListeners.forEach(function(listener) {
                    process.on('uncaughtException', listener);
                });
            }
        """)
    }

    /*
     * Test that `propagateExceptionFinalResort` re-throws the exception on JS.
     *
     * It is checked by setting up an exception handler within JS.
     */
    @Test
    fun testPropagateExceptionFinalResortReThrowsOnNodeJS() = runTest {
        js("""
            globalThis.exceptionCaught = false;
            process.on('uncaughtException', function(e) {
                globalThis.exceptionCaught = true;
            });
        """)
        val job = GlobalScope.launch {
            throw IllegalStateException("My ISE")
        }
        job.join()
        delay(1) // Let the exception be re-thrown and handled.
        val exceptionCaught = js("globalThis.exceptionCaught") as Boolean
        assertTrue(exceptionCaught)
    }
}
Repository: Kotlin/kotlinx.coroutines
File: kotlinx-coroutines-core/wasmJs/test/PropagateExceptionFinalResortTest.kt (Kotlin)

Code before:
package kotlinx.coroutines

import kotlinx.coroutines.testing.TestBase
import kotlin.test.*

fun addUncaughtExceptionHandler() {
    js("""
        globalThis.exceptionCaught = false;
        process.on('uncaughtException', function(e) {
            globalThis.exceptionCaught = true;
        });
    """)
}

fun exceptionCaught(): Boolean = js("globalThis.exceptionCaught")

class PropagateExceptionFinalResortTest : TestBase() {
    @Test
    fun testThrows() = runTest {
        addUncaughtExceptionHandler()
        val job = GlobalScope.launch {
            throw IllegalStateException("My ISE")
        }
        job.join()
        delay(1) // Give time for the error handler to execute
        assertTrue(exceptionCaught())
    }
}

Review comment (suggestion):
Worth marking such things as `private` to avoid calling this function specific to one test from other tests.

Code after:
package kotlinx.coroutines

import kotlinx.coroutines.testing.TestBase
import kotlin.test.*

class PropagateExceptionFinalResortTest : TestBase() {
    @BeforeTest
    private fun addUncaughtExceptionHandler() {
        addUncaughtExceptionHandlerHelper()
    }

    @AfterTest
    private fun removeHandler() {
        removeHandlerHelper()
    }

    /*
     * Test that `propagateExceptionFinalResort` re-throws the exception on Wasm/JS.
     *
     * It is checked by setting up an exception handler within Wasm/JS.
     */
    @Test
    fun testPropagateExceptionFinalResortReThrowsOnWasmJS() = runTest {
        val job = GlobalScope.launch {
            throw IllegalStateException("My ISE")
        }
        job.join()
        delay(1) // Let the exception be re-thrown and handled.
        assertTrue(exceptionCaught())
    }
}