Remove cleaning in RDD#filterWith

Author: tedyu

core/src/main/scala/org/apache/spark/rdd/RDD.scala

@@ -790,10 +790,10 @@ abstract class RDD[T: ClassTag](
   @deprecated("use mapPartitionsWithIndex and filter", "1.0.0")
   def filterWith[A](constructA: Int => A)(p: (T, A) => Boolean): RDD[T] = withScope {
     mapPartitionsWithIndex((index, iter) => {
-      val cleanP = sc.clean(p)
-      val cleanA = sc.clean(constructA)
-      val a = cleanA(index)
-      iter.filter(t => cleanP(t, a))
+      /* val cleanP = sc.clean(p)
+      val cleanA = sc.clean(constructA) */
+      val a = constructA(index)
+      iter.filter(t => p(t, a))
     }, preservesPartitioning = true)
   }
 

core/src/test/scala/org/apache/spark/util/ClosureCleanerSuite.scala

@@ -94,7 +94,7 @@ class ClosureCleanerSuite extends FunSuite {
       expectCorrectException { TestUserClosuresActuallyCleaned.testMapPartitionsWithIndex(rdd) }
       expectCorrectException { TestUserClosuresActuallyCleaned.testMapPartitionsWithContext(rdd) }
       expectCorrectException { TestUserClosuresActuallyCleaned.testFlatMapWith(rdd) }
-      expectCorrectException { TestUserClosuresActuallyCleaned.testFilterWith(rdd) }
+      // expectCorrectException { TestUserClosuresActuallyCleaned.testFilterWith(rdd) }
       expectCorrectException { TestUserClosuresActuallyCleaned.testForEachWith(rdd) }
       expectCorrectException { TestUserClosuresActuallyCleaned.testMapWith(rdd) }
       expectCorrectException { TestUserClosuresActuallyCleaned.testZipPartitions2(rdd) }