From 51809834b432e14beac8a31f9e0076fb7af35c6b Mon Sep 17 00:00:00 2001
From: ChengJie1053 <18033291053@163.com>
Date: Tue, 11 Apr 2023 18:34:47 +0800
Subject: [PATCH] adapt Spark 3.x

---
 rocketmq-spark/pom.xml                                      | 6 +++---
 .../org/apache/spark/sql/rocketmq/RocketMQSource.scala      | 1 -
 .../apache/spark/sql/rocketmq/RocketMQSourceOffset.scala    | 5 ++---
 .../org/apache/spark/sql/rocketmq/RocketMQSourceRDD.scala   | 2 +-
 .../main/scala/org/apache/spark/streaming/RocketMqRDD.scala | 2 +-
 5 files changed, 7 insertions(+), 9 deletions(-)

diff --git a/rocketmq-spark/pom.xml b/rocketmq-spark/pom.xml
index 87810a536..debba0eec 100644
--- a/rocketmq-spark/pom.xml
+++ b/rocketmq-spark/pom.xml
@@ -34,9 +34,9 @@
         <maven.compiler.source>1.8</maven.compiler.source>
         <maven.compiler.target>1.8</maven.compiler.target>
         <rocketmq.version>4.9.4</rocketmq.version>
-        <spark.version>2.3.0</spark.version>
-        <scala.version>2.11.8</scala.version>
-        <scala.binary.version>2.11</scala.binary.version>
+        <spark.version>3.3.1</spark.version>
+        <scala.version>2.12.10</scala.version>
+        <scala.binary.version>2.12</scala.binary.version>
         <commons-lang.version>2.5</commons-lang.version>
     </properties>
 
diff --git a/rocketmq-spark/src/main/scala/org/apache/spark/sql/rocketmq/RocketMQSource.scala b/rocketmq-spark/src/main/scala/org/apache/spark/sql/rocketmq/RocketMQSource.scala
index efc671d7e..b526bf2cb 100644
--- a/rocketmq-spark/src/main/scala/org/apache/spark/sql/rocketmq/RocketMQSource.scala
+++ b/rocketmq-spark/src/main/scala/org/apache/spark/sql/rocketmq/RocketMQSource.scala
@@ -115,7 +115,6 @@ private class RocketMQSource(
           if (content(0) == 'v') {
             val indexOfNewLine = content.indexOf("\n")
             if (indexOfNewLine > 0) {
-              val version = parseVersion(content.substring(0, indexOfNewLine), VERSION)
               RocketMQSourceOffset(SerializedOffset(content.substring(indexOfNewLine + 1)))
             } else {
               throw new IllegalStateException(
diff --git a/rocketmq-spark/src/main/scala/org/apache/spark/sql/rocketmq/RocketMQSourceOffset.scala b/rocketmq-spark/src/main/scala/org/apache/spark/sql/rocketmq/RocketMQSourceOffset.scala
index 7635de38b..c8f702cfe 100644
--- a/rocketmq-spark/src/main/scala/org/apache/spark/sql/rocketmq/RocketMQSourceOffset.scala
+++ b/rocketmq-spark/src/main/scala/org/apache/spark/sql/rocketmq/RocketMQSourceOffset.scala
@@ -25,15 +25,14 @@
 package org.apache.spark.sql.rocketmq
 
 import org.apache.rocketmq.common.message.MessageQueue
+import org.apache.spark.sql.connector.read.streaming.PartitionOffset
 import org.apache.spark.sql.execution.streaming.{Offset, SerializedOffset}
-import org.apache.spark.sql.sources.v2.reader.streaming.{PartitionOffset, Offset => OffsetV2}
-
 /**
  * An [[Offset]] for the [[RocketMQSource]]. This one tracks all partitions of subscribed topics and
  * their offsets.
  */
 private[rocketmq]
-case class RocketMQSourceOffset(queueToOffsets: Map[MessageQueue, Long]) extends OffsetV2 {
+case class RocketMQSourceOffset(queueToOffsets: Map[MessageQueue, Long]) extends Offset {
   override val json = JsonUtils.partitionOffsets(queueToOffsets)
 }
 
diff --git a/rocketmq-spark/src/main/scala/org/apache/spark/sql/rocketmq/RocketMQSourceRDD.scala b/rocketmq-spark/src/main/scala/org/apache/spark/sql/rocketmq/RocketMQSourceRDD.scala
index d30eef1ba..0fcd7bafa 100644
--- a/rocketmq-spark/src/main/scala/org/apache/spark/sql/rocketmq/RocketMQSourceRDD.scala
+++ b/rocketmq-spark/src/main/scala/org/apache/spark/sql/rocketmq/RocketMQSourceRDD.scala
@@ -174,7 +174,7 @@ private[rocketmq] class RocketMQSourceRDD(
         }
       }
       // Release consumer, either by removing it or indicating we're no longer using it
-      context.addTaskCompletionListener { _ =>
+      context.addTaskCompletionListener[Unit] { _ =>
         underlying.closeIfNeeded()
       }
       underlying
diff --git a/rocketmq-spark/src/main/scala/org/apache/spark/streaming/RocketMqRDD.scala b/rocketmq-spark/src/main/scala/org/apache/spark/streaming/RocketMqRDD.scala
index 08a23520a..fc4e46bb5 100644
--- a/rocketmq-spark/src/main/scala/org/apache/spark/streaming/RocketMqRDD.scala
+++ b/rocketmq-spark/src/main/scala/org/apache/spark/streaming/RocketMqRDD.scala
@@ -178,7 +178,7 @@ class RocketMqRDD (
     logDebug(s"Computing topic ${part.topic}, queueId ${part.queueId} " +
       s"offsets ${part.partitionOffsetRanges.mkString(",")}")
 
-    context.addTaskCompletionListener{ context => closeIfNeeded() }
+    context.addTaskCompletionListener[Unit]{ context => closeIfNeeded() }
 
 
     val consumer = if (useConsumerCache) {