It compiles with the Java 8 happy pandas

core/src/main/scala/org/apache/spark/api/java/function/MapPartitionsWithIndexFunction.scala → core/src/main/java/org/apache/spark/api/java/function/MapPartitionsWithIndexFunction.java

@@ -15,16 +15,13 @@
  * limitations under the License.
  */
 
-package org.apache.spark.api.java.function
+package org.apache.spark.api.java.function;
 
-import scala.reflect.ClassTag
-
-import org.apache.spark.api.java.JavaSparkContext.fakeClassTag
+import java.io.Serializable;
 
 /**
  * A function that takes the a partition
  */
-abstract class MapPartitionsWithIndexFunction[A, B] extends Function2[Integer,
-  java.util.Iterator[A], java.util.Iterator[B]] {
-  def elementType() : ClassTag[B] = fakeClassTag
+public interface MapPartitionsWithIndexFunction<A, B> extends Function2<Integer,
+  java.util.Iterator<A>, java.util.Iterator<B>> {
 }

core/src/main/scala/org/apache/spark/api/java/JavaRDDLike.scala

@@ -72,12 +72,12 @@ trait JavaRDDLike[T, This <: JavaRDDLike[T, This]] extends Serializable {
    * Return a new RDD by applying a function to each partition of this RDD, while tracking the index
    * of the original partition.
    */
-  def mapPartitionsWithIndex[R](f: MapPartitionsWithIndexFunction[T, R],
+  def mapPartitionsWithIndex[R: ClassTag](f: MapPartitionsWithIndexFunction[T, R],
     preservesPartitioning: Boolean = false): JavaRDD[R] = {
     import scala.collection.JavaConverters._
-    def fn = (a: Int, b: Iterator[T]) => f.apply(a, asJavaIterator(b)).asScala
-    val newRdd = rdd.mapPartitionsWithIndex(fn, preservesPartitioning)(f.elementType())
-    new JavaRDD(newRdd)(f.elementType())
+    def fn = (a: Int, b: Iterator[T]) => f.call(a, asJavaIterator(b)).asScala
+    val newRdd = rdd.mapPartitionsWithIndex(fn, preservesPartitioning)
+    new JavaRDD(newRdd)(fakeClassTag)
   }
 
   /**