Ensure that all coroutines throwables in core are serializable

integration-testing/build.gradle

@@ -23,6 +23,16 @@ dependencies {
 }
 
 sourceSets {
+    withGuavaTest {
+        kotlin
+        compileClasspath += sourceSets.test.runtimeClasspath
+        runtimeClasspath += sourceSets.test.runtimeClasspath
+
+        dependencies {
+            implementation "org.jetbrains.kotlinx:kotlinx-coroutines-core:$coroutines_version"
+            implementation 'com.google.guava:guava:31.1-jre'
+        }
+    }
     mavenTest {
         kotlin
         compileClasspath += sourceSets.test.runtimeClasspath
@@ -60,6 +70,13 @@ compileDebugAgentTestKotlin {
     }
 }
 
+task withGuavaTest(type: Test) {
+    environment "version", coroutines_version
+    def sourceSet = sourceSets.withGuavaTest
+    testClassesDirs = sourceSet.output.classesDirs
+    classpath = sourceSet.runtimeClasspath
+}
+
 task mavenTest(type: Test) {
     environment "version", coroutines_version
     def sourceSet = sourceSets.mavenTest
@@ -89,5 +106,5 @@ compileTestKotlin {
 }
 
 check {
-    dependsOn([mavenTest, debugAgentTest, coreAgentTest, 'smokeTest:build'])
+    dependsOn([withGuavaTest, mavenTest, debugAgentTest, coreAgentTest, 'smokeTest:build'])
 }

integration-testing/src/withGuavaTest/kotlin/ListAllCoroutineThrowableSubclassesTest.kt

@@ -0,0 +1,43 @@
+/*
+ * Copyright 2016-2022 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines
+
+import com.google.common.reflect.*
+import kotlinx.coroutines.*
+import org.junit.Test
+import kotlin.test.*
+
+class ListAllCoroutineThrowableSubclassesTest {
+
+    /*
+     * These are all the known throwables in kotlinx.coroutines.
+     * If you add one, this test will fail to make
+     * you ensure your exception type is java.io.Serializable.
+     *
+     * We do not have means to check it automatically, so checks are delegated to humans.
+     *
+     * See #3328 for serialization rationale.
+     */
+    private val knownThrowables = setOf(
+        "kotlinx.coroutines.TimeoutCancellationException",
+        "kotlinx.coroutines.JobCancellationException",
+        "kotlinx.coroutines.internal.UndeliveredElementException",
+        "kotlinx.coroutines.CompletionHandlerException",
+        "kotlinx.coroutines.DiagnosticCoroutineContextException",
+        "kotlinx.coroutines.CoroutinesInternalError",
+        "kotlinx.coroutines.channels.ClosedSendChannelException",
+        "kotlinx.coroutines.channels.ClosedReceiveChannelException",
+        "kotlinx.coroutines.flow.internal.ChildCancelledException",
+        "kotlinx.coroutines.flow.internal.AbortFlowException",
+        )
+
+    @Test
+    fun testThrowableSubclassesAreSerializable() {
+        val classes = ClassPath.from(this.javaClass.classLoader)
+            .getTopLevelClassesRecursive("kotlinx.coroutines");
+        val throwables = classes.filter { Throwable::class.java.isAssignableFrom(it.load()) }.map { it.toString() }
+        assertEquals(knownThrowables.sorted(), throwables.sorted())
+    }
+}

kotlinx-coroutines-core/common/src/Timeout.kt

@@ -163,7 +163,7 @@ private class TimeoutCoroutine<U, in T: U>(
  */
 public class TimeoutCancellationException internal constructor(
     message: String,
-    @JvmField internal val coroutine: Job?
+    @JvmField @Transient internal val coroutine: Job?
 ) : CancellationException(message), CopyableThrowable<TimeoutCancellationException> {
     /**
      * Creates a timeout exception with the given message.
@@ -173,7 +173,7 @@ public class TimeoutCancellationException internal constructor(
     internal constructor(message: String) : this(message, null)
 
     // message is never null in fact
-    override fun createCopy(): TimeoutCancellationException? =
+    override fun createCopy(): TimeoutCancellationException =
         TimeoutCancellationException(message ?: "", coroutine).also { it.initCause(this) }
 }
 

kotlinx-coroutines-core/jvm/src/Exceptions.kt

@@ -29,7 +29,7 @@ public actual fun CancellationException(message: String?, cause: Throwable?) : C
 internal actual class JobCancellationException public actual constructor(
     message: String,
     cause: Throwable?,
-    @JvmField internal actual val job: Job
+    @JvmField @Transient internal actual val job: Job
 ) : CancellationException(message), CopyableThrowable<JobCancellationException> {
 
     init {

kotlinx-coroutines-core/jvm/src/flow/internal/FlowExceptions.kt

@@ -8,7 +8,7 @@ import kotlinx.coroutines.*
 import kotlinx.coroutines.flow.*
 
 internal actual class AbortFlowException actual constructor(
-    actual val owner: FlowCollector<*>
+    @JvmField @Transient actual val owner: FlowCollector<*>
 ) : CancellationException("Flow was aborted, no more elements needed") {
 
     override fun fillInStackTrace(): Throwable {

kotlinx-coroutines-core/jvm/test/JobCancellationExceptionSerializerTest.kt

@@ -0,0 +1,42 @@
+/*
+ * Copyright 2016-2022 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines
+
+import org.junit.*
+import java.io.*
+
+
+@Suppress("BlockingMethodInNonBlockingContext")
+class JobCancellationExceptionSerializerTest : TestBase() {
+
+    @Test
+    fun testSerialization() = runTest {
+        try {
+            coroutineScope {
+                expect(1)
+
+                launch {
+                    expect(2)
+                    try {
+                        hang {}
+                    } catch (e: CancellationException) {
+                        throw RuntimeException("RE2", e)
+                    }
+                }
+
+                launch {
+                    expect(3)
+                    throw RuntimeException("RE1")
+                }
+            }
+        } catch (e: Throwable) {
+            // Should not fail
+            ObjectOutputStream(ByteArrayOutputStream()).use {
+                it.writeObject(e)
+            }
+            finish(4)
+        }
+    }
+}