From 1d3cd065724f3bd4f5cab90f95b5d5c8d0b389ea Mon Sep 17 00:00:00 2001
From: Yungthuis <36870105+Yungthuis@users.noreply.github.com>
Date: Thu, 8 Jul 2021 13:31:38 +0800
Subject: [PATCH] [HUDI-2134]Add generics to avoif forced conversion in
 BaseSparkCommitActionExecutor#partition (#3232)

---
 .../action/commit/BaseSparkCommitActionExecutor.java   | 10 +++++-----
 1 file changed, 5 insertions(+), 5 deletions(-)

diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/BaseSparkCommitActionExecutor.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/BaseSparkCommitActionExecutor.java
index eceac7bd1..c8b4fff50 100644
--- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/BaseSparkCommitActionExecutor.java
+++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/BaseSparkCommitActionExecutor.java
@@ -188,16 +188,16 @@ public abstract class BaseSparkCommitActionExecutor<T extends HoodieRecordPayloa
   }
 
   private JavaRDD<HoodieRecord<T>> partition(JavaRDD<HoodieRecord<T>> dedupedRecords, Partitioner partitioner) {
-    JavaPairRDD<Tuple2, HoodieRecord<T>> mappedRDD = dedupedRecords.mapToPair(
+    JavaPairRDD<Tuple2<HoodieKey, Option<HoodieRecordLocation>>, HoodieRecord<T>> mappedRDD = dedupedRecords.mapToPair(
         record -> new Tuple2<>(new Tuple2<>(record.getKey(), Option.ofNullable(record.getCurrentLocation())), record));
 
-    JavaPairRDD<Tuple2, HoodieRecord<T>> partitionedRDD;
+    JavaPairRDD<Tuple2<HoodieKey, Option<HoodieRecordLocation>>, HoodieRecord<T>> partitionedRDD;
     if (table.requireSortedRecords()) {
       // Partition and sort within each partition as a single step. This is faster than partitioning first and then
       // applying a sort.
-      Comparator<Tuple2> comparator = (Comparator<Tuple2> & Serializable)(t1, t2) -> {
-        HoodieKey key1 = (HoodieKey) t1._1;
-        HoodieKey key2 = (HoodieKey) t2._1;
+      Comparator<Tuple2<HoodieKey, Option<HoodieRecordLocation>>> comparator = (Comparator<Tuple2<HoodieKey, Option<HoodieRecordLocation>>> & Serializable)(t1, t2) -> {
+        HoodieKey key1 = t1._1;
+        HoodieKey key2 = t2._1;
         return key1.getRecordKey().compareTo(key2.getRecordKey());
       };