Add Java version and additional doc

Author: sryza

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

@@ -758,6 +758,32 @@ class JavaPairRDD[K, V](val rdd: RDD[(K, V)])
     rdd.saveAsHadoopDataset(conf)
   }
 
+  /**
+   * Repartition the RDD according to the given partitioner and, within each resulting partition,
+   * sort records by their keys.
+   *
+   * This is more efficient than calling `repartition` and then sorting within each partition
+   * because it can push the sorting down into the shuffle machinery.
+   */
+  def repartitionAndSortWithinPartition(partitioner: Partitioner): JavaPairRDD[K, V] = {
+    val comp = com.google.common.collect.Ordering.natural().asInstanceOf[Comparator[K]]
+    repartitionAndSortWithinPartition(partitioner, comp)
+  }
+
+  /**
+   * Repartition the RDD according to the given partitioner and, within each resulting partition,
+   * sort records by their keys.
+   *
+   * This is more efficient than calling `repartition` and then sorting within each partition
+   * because it can push the sorting down into the shuffle machinery.
+   */
+  def repartitionAndSortWithinPartition(partitioner: Partitioner, comp: Comparator)
+    : JavaPairRDD[K, V] = {
+    implicit val ordering = comp // Allow implicit conversion of Comparator to Ordering.
+    fromRDD(
+      new OrderedRDDFunctions[K, V, (K, V)](rdd).repartitionAndSortWithinPartition(partitioner))
+  }
+
   /**
    * Sort the RDD by key, so that each partition contains a sorted range of the elements in
    * ascending order. Calling `collect` or `save` on the resulting RDD will return or output an

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

@@ -68,11 +68,13 @@ class OrderedRDDFunctions[K : Ordering : ClassTag,
   /**
    * Repartition the RDD according to the given partitioner and, within each resulting partition,
    * sort records by their keys.
+   *
+   * This is more efficient than calling `repartition` and then sorting within each partition
+   * because it can push the sorting down into the shuffle machinery.
    */
-  def repartitionAndSortWithinPartition(partitioner: Partitioner, ascending: Boolean = true)
+  def repartitionAndSortWithinPartition(partitioner: Partitioner)
       : RDD[(K, V)] = {
-    new ShuffledRDD[K, V, V](self, partitioner)
-      .setKeyOrdering(if (ascending) ordering else ordering.reverse)
+    new ShuffledRDD[K, V, V](self, partitioner).setKeyOrdering(ordering)
   }
 
 }