Enable -Xsource:3-cross compiler flag

This not only eases potential future Scala 3 migration but also make the compiler stricter with features that have proven to be warts.

Author: joan38

common/utils/src/main/scala/org/apache/spark/internal/config/ConfigBuilder.scala

@@ -94,7 +94,7 @@ private[spark] class TypedConfigBuilder[T](
   import ConfigHelpers._
 
   def this(parent: ConfigBuilder, converter: String => T) = {
-    this(parent, converter, { v: T => v.toString })
+    this(parent, converter, (v: T) => v.toString)
   }
 
   /** Apply a transformation to the user-provided values of the config entry. */

connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaOffsetReaderAdmin.scala

@@ -336,7 +336,7 @@ private[kafka010] class KafkaOffsetReaderAdmin(
         //    admin has a much bigger chance to hit KAFKA-7703 like issues.
         var incorrectOffsets: Seq[(TopicPartition, Long, Long)] = Nil
         var attempt = 0
-        do {
+        while ({
           partitionOffsets = listOffsets(admin, listOffsetsParams)
           attempt += 1
 
@@ -349,7 +349,8 @@ private[kafka010] class KafkaOffsetReaderAdmin(
               Thread.sleep(offsetFetchAttemptIntervalMs)
             }
           }
-        } while (incorrectOffsets.nonEmpty && attempt < maxOffsetFetchAttempts)
+          incorrectOffsets.nonEmpty && attempt < maxOffsetFetchAttempts
+        }) ()
 
         logDebug(s"Got latest offsets for partitions: $partitionOffsets")
         partitionOffsets

connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaOffsetReaderConsumer.scala

@@ -385,7 +385,7 @@ private[kafka010] class KafkaOffsetReaderConsumer(
         // - Avoid calling `consumer.poll(0)` which may cause KAFKA-7703.
         var incorrectOffsets: Seq[(TopicPartition, Long, Long)] = Nil
         var attempt = 0
-        do {
+        while ({
           consumer.seekToEnd(partitions)
           partitionOffsets = partitions.asScala.map(p => p -> consumer.position(p)).toMap
           attempt += 1
@@ -399,7 +399,8 @@ private[kafka010] class KafkaOffsetReaderConsumer(
               Thread.sleep(offsetFetchAttemptIntervalMs)
             }
           }
-        } while (incorrectOffsets.nonEmpty && attempt < maxOffsetFetchAttempts)
+          incorrectOffsets.nonEmpty && attempt < maxOffsetFetchAttempts
+        }) ()
 
         logDebug(s"Got latest offsets for partition : $partitionOffsets")
         partitionOffsets

connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/consumer/FetchedDataPool.scala

@@ -151,7 +151,7 @@ private[consumer] class FetchedDataPool(
   private def removeIdleFetchedData(): Unit = synchronized {
     val now = clock.getTimeMillis()
     val maxAllowedReleasedTimestamp = now - minEvictableIdleTimeMillis
-    cache.values.foreach { p: CachedFetchedDataList =>
+    cache.values.foreach { (p: CachedFetchedDataList) =>
       val expired = p.filter { q =>
         !q.inUse && q.lastReleasedTimestamp < maxAllowedReleasedTimestamp
       }

connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchSourceSuite.scala

@@ -2362,7 +2362,7 @@ abstract class KafkaSourceSuiteBase extends KafkaSourceTest {
       AddKafkaData(Set(topic), 30, 31, 32), // Add data when stream is stopped
       StartStream(),
       CheckAnswer(-21, -22, -11, -12, 2, 12, 20, 21, 22, 30, 31, 32), // Should get the added data
-      AssertOnQuery("Add partitions") { query: StreamExecution =>
+      AssertOnQuery("Add partitions") { (query: StreamExecution) =>
         if (addPartitions) setTopicPartitions(topic, 10, query)
         true
       },
@@ -2413,7 +2413,7 @@ abstract class KafkaSourceSuiteBase extends KafkaSourceTest {
       AddKafkaData(Set(topic), 30, 31, 32), // Add data when stream is stopped
       StartStream(),
       CheckAnswer(-21, -22, -11, -12, 2, 12, 23, 24, 30, 31, 32), // Should get the added data
-      AssertOnQuery("Add partitions") { query: StreamExecution =>
+      AssertOnQuery("Add partitions") { (query: StreamExecution) =>
         if (addPartitions) setTopicPartitions(topic, 10, query)
         true
       },
@@ -2581,7 +2581,7 @@ abstract class KafkaSourceSuiteBase extends KafkaSourceTest {
       CheckAnswer(2, 3, 4, 5, 6, 7), // Should get the added data
       AddKafkaData(Set(topic), 7, 8),
       CheckAnswer(2, 3, 4, 5, 6, 7, 8, 9),
-      AssertOnQuery("Add partitions") { query: StreamExecution =>
+      AssertOnQuery("Add partitions") { (query: StreamExecution) =>
         if (addPartitions) setTopicPartitions(topic, 10, query)
         true
       },
@@ -2622,7 +2622,7 @@ abstract class KafkaSourceSuiteBase extends KafkaSourceTest {
       AddKafkaData(Set(topic), 7, 8),
       StartStream(),
       CheckAnswer(2, 3, 4, 5, 6, 7, 8, 9),
-      AssertOnQuery("Add partitions") { query: StreamExecution =>
+      AssertOnQuery("Add partitions") { (query: StreamExecution) =>
         if (addPartitions) setTopicPartitions(topic, 10, query)
         true
       },

connector/kafka-0-10/src/main/scala/org/apache/spark/streaming/kafka010/DirectKafkaInputDStream.scala

@@ -112,7 +112,7 @@ private[spark] class DirectKafkaInputDStream[K, V](
 
   private[streaming] override def name: String = s"Kafka 0.10 direct stream [$id]"
 
-  protected[streaming] override val checkpointData =
+  protected[streaming] override val checkpointData: DirectKafkaInputDStreamCheckpointData =
     new DirectKafkaInputDStreamCheckpointData
 
 

connector/kinesis-asl/src/test/scala/org/apache/spark/streaming/kinesis/KinesisTestUtils.scala

@@ -176,10 +176,11 @@ private[kinesis] class KinesisTestUtils(streamShardCount: Int = 2) extends Loggi
 
   private def findNonExistentStreamName(): String = {
     var testStreamName: String = null
-    do {
+    while ({
       Thread.sleep(TimeUnit.SECONDS.toMillis(describeStreamPollTimeSeconds))
       testStreamName = s"KinesisTestUtils-${math.abs(Random.nextLong())}"
-    } while (describeStream(testStreamName).nonEmpty)
+      describeStream(testStreamName).nonEmpty
+    }) ()
     testStreamName
   }
 

connector/protobuf/src/main/scala/org/apache/spark/sql/protobuf/ProtobufSerializer.scala

@@ -322,7 +322,7 @@ private[sql] class ProtobufSerializer(
       .unzip
 
     val numFields = catalystStruct.length
-    row: InternalRow =>
+    (row: InternalRow) =>
       val result = DynamicMessage.newBuilder(descriptor)
       var i = 0
       while (i < numFields) {

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

@@ -29,7 +29,7 @@ import org.apache.spark.rdd.HadoopRDD
 
 @DeveloperApi
 class JavaHadoopRDD[K, V](rdd: HadoopRDD[K, V])
-    (implicit override val kClassTag: ClassTag[K], implicit override val vClassTag: ClassTag[V])
+    (implicit override val kClassTag: ClassTag[K], override val vClassTag: ClassTag[V])
   extends JavaPairRDD[K, V](rdd) {
 
   /** Maps over a partition, providing the InputSplit that was used as the base of the partition. */

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

@@ -29,7 +29,7 @@ import org.apache.spark.rdd.NewHadoopRDD
 
 @DeveloperApi
 class JavaNewHadoopRDD[K, V](rdd: NewHadoopRDD[K, V])
-    (implicit override val kClassTag: ClassTag[K], implicit override val vClassTag: ClassTag[V])
+    (implicit override val kClassTag: ClassTag[K], override val vClassTag: ClassTag[V])
   extends JavaPairRDD[K, V](rdd) {
 
   /** Maps over a partition, providing the InputSplit that was used as the base of the partition. */