Do not throw exception if object to clean is not closure

This breaks a valid use case where the user code passes in a case
class into `map`. See ml.NormalizerSuite.

Author: Andrew Or

core/src/main/scala/org/apache/spark/util/ClosureCleaner.scala

@@ -180,7 +180,8 @@ private[spark] object ClosureCleaner extends Logging {
       accessedFields: Map[Class[_], Set[String]]): Unit = {
 
     if (!isClosure(func.getClass)) {
-      throw new IllegalArgumentException("Expected a closure; got " + func.getClass.getName)
+      logWarning("Expected a closure; got " + func.getClass.getName)
+      return
     }
 
     // TODO: clean all inner closures first. This requires us to find the inner objects.

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

@@ -63,18 +63,6 @@ class ClosureCleanerSuite extends FunSuite {
     val result = TestObjectWithNestedReturns.run()
     assert(result === 1)
   }
-
-  test("should clean only closures") {
-    withSpark(new SparkContext("local", "test")) { sc =>
-      val rdd = sc.makeRDD(1 to 100)
-      intercept[IllegalArgumentException] { sc.clean(new Integer(1)) }
-      intercept[IllegalArgumentException] { sc.clean("not a closure") }
-      intercept[IllegalArgumentException] { sc.clean(rdd) }
-      sc.clean(() => new Integer(1))
-      sc.clean(() => "part of a closure")
-      sc.clean(() => rdd)
-    }
-  }
 }
 
 // A non-serializable class we create in closures to make sure that we aren't

streaming/src/test/scala/org/apache/spark/streaming/ReceiverSuite.scala

@@ -256,8 +256,8 @@ class ReceiverSuite extends TestSuiteBase with Timeouts with Serializable {
     }
 
     withStreamingContext(new StreamingContext(sparkConf, batchDuration)) { ssc =>
-      val receiver1 = ssc.sparkContext.clean(new FakeReceiver(sendData = true))
-      val receiver2 = ssc.sparkContext.clean(new FakeReceiver(sendData = true))
+      val receiver1 = new FakeReceiver(sendData = true)
+      val receiver2 = new FakeReceiver(sendData = true)
       val receiverStream1 = ssc.receiverStream(receiver1)
       val receiverStream2 = ssc.receiverStream(receiver2)
       receiverStream1.register()