[SPARK-32873][BUILD] Fix code which causes error when build with sbt and Scala 2.13

external/kafka-0-10/src/main/scala/org/apache/spark/streaming/kafka010/KafkaRDD.scala

@@ -248,7 +248,7 @@ private class KafkaRDDIterator[K, V](
     }
   }
 
-  override def hasNext(): Boolean = requestOffset < part.untilOffset
+  override def hasNext: Boolean = requestOffset < part.untilOffset
 
   override def next(): ConsumerRecord[K, V] = {
     if (!hasNext) {
@@ -291,7 +291,7 @@ private class CompactedKafkaRDDIterator[K, V](
 
   private var okNext: Boolean = true
 
-  override def hasNext(): Boolean = okNext
+  override def hasNext: Boolean = okNext
 
   override def next(): ConsumerRecord[K, V] = {
     if (!hasNext) {

mllib-local/src/main/scala/org/apache/spark/ml/linalg/Vectors.scala

@@ -786,7 +786,7 @@ class SparseVector @Since("2.0.0") (
 
       override def hasNext: Boolean = i < localSize
 
-      override def next: (Int, Double) = {
+      override def next(): (Int, Double) = {
         val v = if (i == k) {
           j += 1
           k = if (j < localNumActives) localIndices(j) else -1

mllib/src/main/scala/org/apache/spark/mllib/feature/PCA.scala

@@ -124,9 +124,9 @@ private[feature] object PCAUtil {
   // 6e541be066d547a097f5089165cd7c38c3ca276d/math/src/main/scala/breeze/linalg/
   // functions/svd.scala#L87
   def memoryCost(k: Int, numFeatures: Int): Long = {
-    3L * math.min(k, numFeatures) * math.min(k, numFeatures)
-    + math.max(math.max(k, numFeatures), 4L * math.min(k, numFeatures)
-    * math.min(k, numFeatures) + 4L * math.min(k, numFeatures))
+    3L * math.min(k, numFeatures) * math.min(k, numFeatures) +
+      math.max(math.max(k, numFeatures), 4L * math.min(k, numFeatures) *
+      math.min(k, numFeatures) + 4L * math.min(k, numFeatures))
   }
 
 }

mllib/src/main/scala/org/apache/spark/mllib/linalg/Vectors.scala

@@ -984,7 +984,7 @@ class SparseVector @Since("1.0.0") (
 
       override def hasNext: Boolean = i < localSize
 
-      override def next: (Int, Double) = {
+      override def next(): (Int, Double) = {
         val v = if (i == k) {
           j += 1
           k = if (j < localNumActives) localIndices(j) else -1

sql/core/src/main/scala/org/apache/spark/sql/execution/command/commands.scala

@@ -78,7 +78,7 @@ case class ExecutedCommandExec(cmd: RunnableCommand) extends LeafExecNode {
 
   override def executeCollect(): Array[InternalRow] = sideEffectResult.toArray
 
-  override def executeToIterator: Iterator[InternalRow] = sideEffectResult.toIterator
+  override def executeToIterator(): Iterator[InternalRow] = sideEffectResult.toIterator
 
   override def executeTake(limit: Int): Array[InternalRow] = sideEffectResult.take(limit).toArray
 
@@ -119,7 +119,7 @@ case class DataWritingCommandExec(cmd: DataWritingCommand, child: SparkPlan)
 
   override def executeCollect(): Array[InternalRow] = sideEffectResult.toArray
 
-  override def executeToIterator: Iterator[InternalRow] = sideEffectResult.toIterator
+  override def executeToIterator(): Iterator[InternalRow] = sideEffectResult.toIterator
 
   override def executeTake(limit: Int): Array[InternalRow] = sideEffectResult.take(limit).toArray
 

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2CommandExec.scala

@@ -44,7 +44,7 @@ abstract class V2CommandExec extends SparkPlan {
    */
   override def executeCollect(): Array[InternalRow] = result.toArray
 
-  override def executeToIterator: Iterator[InternalRow] = result.toIterator
+  override def executeToIterator(): Iterator[InternalRow] = result.toIterator
 
   override def executeTake(limit: Int): Array[InternalRow] = result.take(limit).toArray
 

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/jdbc/JDBCTableCatalog.scala

@@ -61,7 +61,7 @@ class JDBCTableCatalog extends TableCatalog with Logging {
         .getTables(null, schemaPattern, "%", Array("TABLE"));
       new Iterator[Identifier] {
         def hasNext = rs.next()
-        def next = Identifier.of(namespace, rs.getString("TABLE_NAME"))
+        def next() = Identifier.of(namespace, rs.getString("TABLE_NAME"))
       }.toArray
     }
   }

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/SymmetricHashJoinStateManager.scala

@@ -171,7 +171,7 @@ class SymmetricHashJoinStateManager(
         return null
       }
 
-      override def close: Unit = {}
+      override def close(): Unit = {}
     }
   }
 
@@ -280,7 +280,7 @@ class SymmetricHashJoinStateManager(
         return reusedRet.withNew(currentKey, currentValue.value, currentValue.matched)
       }
 
-      override def close: Unit = {}
+      override def close(): Unit = {}
     }
   }
 

streaming/src/main/scala/org/apache/spark/streaming/rdd/WriteAheadLogBackedBlockRDD.scala

@@ -96,7 +96,7 @@ class WriteAheadLogBackedBlockRDD[T: ClassTag](
   @transient private val hadoopConfig = sc.hadoopConfiguration
   private val broadcastedHadoopConf = new SerializableConfiguration(hadoopConfig)
 
-  override def isValid(): Boolean = true
+  override def isValid: Boolean = true
 
   override def getPartitions: Array[Partition] = {
     assertValid()

streaming/src/main/scala/org/apache/spark/streaming/receiver/ReceivedBlockHandler.scala

@@ -234,7 +234,7 @@ private[streaming] class CountingIterator[T](iterator: Iterator[T]) extends Iter
 
    private def isFullyConsumed: Boolean = !iterator.hasNext
 
-   def hasNext(): Boolean = iterator.hasNext
+   def hasNext: Boolean = iterator.hasNext
 
    def count(): Option[Long] = {
      if (isFullyConsumed) Some(_count) else None