Add one more test case: indirectly nested closure. This test case is known to fail for now. Fix will be in a another update commit.

Directly nested closures:
val outerClosure = (x: Int) => {
  val innerClosure = (y: Int) => { ... } // inner closure -> outer closure
}

Indirectly nested closures:
val outerClosure = (x: Int) => {
  class InnerFoo {
    val innerClosure = (y: Int) => { ... } // inner closure -> inner class -> outer closure
  }
  val foo = new InnerFoo
  foo.innerClosure
}

Author: rednaxelafx

repl/src/test/scala/org/apache/spark/repl/SingletonReplSuite.scala

@@ -381,6 +381,12 @@ class SingletonReplSuite extends SparkFunSuite {
   }
 
   test("SPARK-31399: should clone+clean line object w/ non-serializable state in ClosureCleaner") {
+    // Test ClosureCleaner when a closure captures the enclosing `this` REPL line object, and that
+    // object contains an unused non-serializable field.
+    // Specifically, the closure in this test case contains a directly nested closure, and the
+    // capture is triggered by the inner closure.
+    // `ns` should be nulled out, but `topLevelValue` should stay intact.
+
     // Can't use :paste mode because PipedOutputStream/PipedInputStream doesn't work well with the
     // EOT control character (i.e. Ctrl+D).
     // Just write things on a single line to emulate :paste mode.
@@ -402,7 +408,36 @@ class SingletonReplSuite extends SparkFunSuite {
         |}
         |val r = sc.parallelize(0 to 2).map(closure).collect
       """.stripMargin)
-    assertContains("r: Array[scala.collection.immutable.IndexedSeq[String]] = Array(Vector", output)
+    assertContains("r: Array[scala.collection.immutable.IndexedSeq[String]] = " +
+      "Array(Vector(), Vector(1someValue), Vector(1someValue, 1someValue, 2someValue))", output)
+    assertDoesNotContain("Exception", output)
+  }
+
+  test("SPARK-31399: ClosureCleaner should discover indirectly nested closure in inner class") {
+    // Similar to the previous test case, but with indirect closure nesting instead.
+    // There's still nested closures involved, but the inner closure is indirectly nested in the
+    // outer closure, with a level of inner class in between them.
+    // This changes how the inner closure references/captures the outer closure/enclosing `this`
+    // REPL line object, and covers a different code path in inner closure discovery.
+
+    // `ns` should be nulled out, but `topLevelValue` should stay intact.
+
+    val output = runInterpreter(
+      """
+        |class NotSerializableClass(val x: Int)
+        |val ns = new NotSerializableClass(42); val topLevelValue = "someValue"; val closure =
+        |(j: Int) => {
+        |  class InnerFoo {
+        |    val innerClosure = (x: Int) => (1 to x).map { y => y + topLevelValue }
+        |  }
+        |  val innerFoo = new InnerFoo
+        |  (1 to j).flatMap(innerFoo.innerClosure)
+        |}
+        |val r = sc.parallelize(0 to 2).map(closure).collect
+      """.stripMargin)
+    assertContains("r: Array[scala.collection.immutable.IndexedSeq[String]] = " +
+       "Array(Vector(), Vector(1someValue), Vector(1someValue, 1someValue, 2someValue))", output)
+    assertDoesNotContain("Array(Vector(), Vector(1null), Vector(1null, 1null, 2null)", output)
     assertDoesNotContain("Exception", output)
   }