From a6890318ea01666ead36b02668eba9afc43902e1 Mon Sep 17 00:00:00 2001
From: sksamuel <sam@sksamuel.com>
Date: Mon, 25 Jan 2016 19:15:37 +0000
Subject: [PATCH] Removed spark repl detection code from closure cleaner

---
 .../flink/api/scala/ClosureCleaner.scala      | 48 +++++--------------
 1 file changed, 11 insertions(+), 37 deletions(-)

diff --git a/flink-scala/src/main/scala/org/apache/flink/api/scala/ClosureCleaner.scala b/flink-scala/src/main/scala/org/apache/flink/api/scala/ClosureCleaner.scala
index 9b0d62e3725ed..2c6d3cd84d52b 100644
--- a/flink-scala/src/main/scala/org/apache/flink/api/scala/ClosureCleaner.scala
+++ b/flink-scala/src/main/scala/org/apache/flink/api/scala/ClosureCleaner.scala
@@ -126,18 +126,9 @@ object ClosureCleaner {
       LOG.debug("accessedFields: " + accessedFields)
     }
 
-    val inInterpreter = {
-      try {
-        val interpClass = Class.forName("spark.repl.Main")
-        interpClass.getMethod("interp").invoke(null) != null
-      } catch {
-        case _: ClassNotFoundException => true
-      }
-    }
-
     var outerPairs: List[(Class[_], AnyRef)] = (outerClasses zip outerObjects).reverse
     var outer: AnyRef = null
-    if (outerPairs.size > 0 && !isClosure(outerPairs.head._1)) {
+    if (outerPairs.nonEmpty && !isClosure(outerPairs.head._1)) {
       // The closure is ultimately nested inside a class; keep the object of that
       // class without cloning it since we don't want to clone the user's objects.
       outer = outerPairs.head._2
@@ -146,7 +137,7 @@ object ClosureCleaner {
     // Clone the closure objects themselves, nulling out any fields that are not
     // used in the closure we're working on or any of its inner closures.
     for ((cls, obj) <- outerPairs) {
-      outer = instantiateClass(cls, outer, inInterpreter)
+      outer = instantiateClass(cls, outer)
       for (fieldName <- accessedFields(cls)) {
         val field = cls.getDeclaredField(fieldName)
         field.setAccessible(true)
@@ -180,35 +171,18 @@ object ClosureCleaner {
     }
   }
 
-  private def instantiateClass(cls: Class[_], outer: AnyRef, inInterpreter: Boolean): AnyRef = {
+  private def instantiateClass(cls: Class[_], outer: AnyRef): AnyRef = {
     if (LOG.isDebugEnabled) {
       LOG.debug("Creating a " + cls + " with outer = " + outer)
     }
-    if (!inInterpreter) {
-      // This is a bona fide closure class, whose constructor has no effects
-      // other than to set its fields, so use its constructor
-      val cons = cls.getConstructors()(0)
-      val params = cons.getParameterTypes.map(createNullValue).toArray
-      if (outer != null) {
-        params(0) = outer // First param is always outer object
-      }
-      cons.newInstance(params: _*).asInstanceOf[AnyRef]
-    } else {
-      // Use reflection to instantiate object without calling constructor
-      val rf = sun.reflect.ReflectionFactory.getReflectionFactory()
-      val parentCtor = classOf[java.lang.Object].getDeclaredConstructor()
-      val newCtor = rf.newConstructorForSerialization(cls, parentCtor)
-      val obj = newCtor.newInstance().asInstanceOf[AnyRef]
-      if (outer != null) {
-        if (LOG.isDebugEnabled) {
-          LOG.debug("3: Setting $outer on " + cls + " to " + outer)
-        }
-        val field = cls.getDeclaredField("$outer")
-        field.setAccessible(true)
-        field.set(obj, outer)
-      }
-      obj
+    // This is a bona fide closure class, whose constructor has no effects
+    // other than to set its fields, so use its constructor
+    val cons = cls.getConstructors()(0)
+    val params = cons.getParameterTypes.map(createNullValue)
+    if (outer != null) {
+      params(0) = outer // First param is always outer object
     }
+    cons.newInstance(params: _*).asInstanceOf[AnyRef]
   }
 
   /** Copy all data from an InputStream to an OutputStream */
@@ -311,7 +285,7 @@ private[flink] class InnerClosureFinder(output: Set[Class[_]]) extends ClassVisi
       override def visitMethodInsn(op: Int, owner: String, name: String,
                                    desc: String) {
         val argTypes = Type.getArgumentTypes(desc)
-        if (op == INVOKESPECIAL && name == "<init>" && argTypes.length > 0
+        if (op == INVOKESPECIAL && name == "<init>" && argTypes.nonEmpty
           && argTypes(0).toString.startsWith("L") // is it an object?
           && argTypes(0).getInternalName == myName) {
           output += Class.forName(