Supports user defined consumer groups

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>
 

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(

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)
 }
 

rocketmq-spark/src/main/scala/org/apache/spark/sql/rocketmq/RocketMQSourceProvider.scala

@@ -77,13 +77,16 @@ class RocketMQSourceProvider extends DataSourceRegister
     // Each running query should use its own group id. Otherwise, the query may be only assigned
     // partial data since RocketMQ will assign partitions to multiple consumers having the same group
     // id. Hence, we should generate a unique id for each query.
-    val uniqueGroupId = s"spark-rocketmq-source-${UUID.randomUUID}-${metadataPath.hashCode.toHexString}"
+    var uniqueGroupId = s"spark-rocketmq-source-${UUID.randomUUID}-${metadataPath.hashCode.toHexString}"
 
     val caseInsensitiveParams = parameters.map { case (k, v) => (k.toLowerCase(Locale.ROOT), v) }
 
     val startingStreamOffsets = RocketMQSourceProvider.getRocketMQOffsetRangeLimit(caseInsensitiveParams,
       RocketMQConf.CONSUMER_OFFSET, LatestOffsetRangeLimit)
 
+    if (caseInsensitiveParams.contains(RocketMQConf.CONSUMER_GROUP)) {
+      uniqueGroupId = caseInsensitiveParams.get(RocketMQConf.CONSUMER_GROUP).get
+    }
     val offsetReader = new RocketMQOffsetReader(
       paramsForDriver(caseInsensitiveParams),
       parameters,
@@ -252,10 +255,10 @@ object RocketMQSourceProvider extends Logging {
   }
 
   def paramsForDriver(specifiedRocketMQParams: Map[String, String]): ju.Map[String, String] = {
-    if (specifiedRocketMQParams.contains(RocketMQConf.CONSUMER_GROUP)) {
-      throw new IllegalArgumentException(
-        s"Option '${RocketMQConf.CONSUMER_GROUP}' can not be specified")
-    }
+//    if (specifiedRocketMQParams.contains(RocketMQConf.CONSUMER_GROUP)) {
+//      throw new IllegalArgumentException(
+//        s"Option '${RocketMQConf.CONSUMER_GROUP}' can not be specified")
+//    }
     ConfigUpdater("source", specifiedRocketMQParams)
         // Set to "earliest" to avoid exceptions. However, RocketMQSource will fetch the initial
         // offsets by itself instead of counting on RocketMQConsumer.
@@ -268,10 +271,10 @@ object RocketMQSourceProvider extends Logging {
   def paramsForExecutors(
       specifiedRocketMQParams: Map[String, String],
       uniqueGroupId: String): ju.Map[String, String] = {
-    if (specifiedRocketMQParams.contains(RocketMQConf.CONSUMER_GROUP)) {
-      throw new IllegalArgumentException(
-        s"Option '${RocketMQConf.CONSUMER_GROUP}' can not be specified")
-    }
+//    if (specifiedRocketMQParams.contains(RocketMQConf.CONSUMER_GROUP)) {
+//      throw new IllegalArgumentException(
+//        s"Option '${RocketMQConf.CONSUMER_GROUP}' can not be specified")
+//    }
     ConfigUpdater("executor", specifiedRocketMQParams)
         // So that consumers in executors do not mess with any existing group id
         .set(RocketMQConf.CONSUMER_GROUP, s"$uniqueGroupId-executor")

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

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) {