Spark 3.5.1

pom.xml

@@ -27,8 +27,8 @@
 
 
   <properties>
-    <spark.version>3.4.1</spark.version>
-    <scala.binary.version>2.12</scala.binary.version>
+    <spark.version>3.5.1</spark.version>
+    <scala.binary.version>2.13</scala.binary.version>
     <fasterxml.jackson.version>2.14.3</fasterxml.jackson.version>
     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>

src/main/scala/org/apache/spark/sql/connector/kinesis/client/KinesisClientConsumerImpl.scala

@@ -265,7 +265,7 @@ case class KinesisClientConsumerImpl(
   override def getShards: Seq[Shard] = {
     val shards = listShards
     logDebug(s"List shards in Kinesis Stream:  ${shards}")
-    shards
+    shards.toSeq
   }
 
   override def close(): Unit = synchronized {

src/main/scala/org/apache/spark/sql/connector/kinesis/metadata/dynamodb/DynamodbMetadataCommitter.scala

@@ -311,7 +311,7 @@ class DynamodbMetadataCommitter[T <: AnyRef : ClassTag](
       queryDataByBatchId(batchId, fromDynamoItem)
     }
     logDebug(s"get result size ${result.length} from table ${table} for batchId ${batchId}")
-    result
+    result.toSeq
   }
 
   override def purgeBefore(thresholdBatchId: Long): Unit = {
@@ -515,7 +515,7 @@ class DynamodbMetadataCommitter[T <: AnyRef : ClassTag](
         }
       }
     }
-    result
+    result.toSeq
   }
 
   // limit: specify the approximate number of items to return. It is a hint not a hard limit.

src/main/scala/org/apache/spark/sql/connector/kinesis/retrieval/RecordBatch.scala

@@ -42,12 +42,12 @@ case class RecordBatch (
         rawRecords.asJava,
         streamShard.shard.hashKeyRange().startingHashKey(),
         streamShard.shard.hashKeyRange().endingHashKey(),
-        millisBehindLatest).asScala
+        millisBehindLatest).asScala.toSeq
     } else {
       aggregatorUtil.deaggregate(
         rawRecords.asJava,
         millisBehindLatest
-      ).asScala
+      ).asScala.toSeq
     }
   }
 

src/main/scala/org/apache/spark/sql/connector/kinesis/retrieval/efo/EfoRecordBatchPublisher.scala

@@ -54,7 +54,7 @@ class EfoRecordBatchPublisher (
     logDebug(s"EfoRecordBatchPublisher runProcessLoop on ${streamShard}, startingPosition: ${nextStartingPosition}")
     val eventConsumer = new Consumer[SubscribeToShardEvent]() {
       override def accept(event: SubscribeToShardEvent): Unit = {
-        val recordBatch = RecordBatch (event.records.asScala, streamShard, event.millisBehindLatest)
+        val recordBatch = RecordBatch(event.records.asScala.toSeq, streamShard, event.millisBehindLatest)
         val sequenceNumber = recordBatchConsumer.accept (recordBatch)
         nextStartingPosition = getNextStartingPosition (sequenceNumber, nextStartingPosition)
       }}

src/main/scala/org/apache/spark/sql/connector/kinesis/retrieval/polling/PollingRecordBatchPublisher.scala

@@ -79,7 +79,7 @@ class PollingRecordBatchPublisher(
     if (nextShardItr == null) return COMPLETE
 
     val result = getRecords(nextShardItr, maxNumberOfRecords)
-    val recordBatch = RecordBatch(result.records().asScala, streamShard, result.millisBehindLatest)
+    val recordBatch = RecordBatch(result.records().asScala.toSeq, streamShard, result.millisBehindLatest)
     val latestSequenceNumber = consumer.accept(recordBatch)
     nextStartingPosition = getNextStartingPosition(latestSequenceNumber, nextStartingPosition)
     nextShardItr = result.nextShardIterator

src/test/scala/org/apache/spark/sql/connector/kinesis/KinesisTestUtils.scala

@@ -105,7 +105,7 @@ class KinesisTestUtils(streamShardCount: Int = 2) extends Logging {
   }
 
   def getShards(): Seq[Shard] = {
-    kinesisClient.describeStream(_streamName).getStreamDescription.getShards.asScala
+    kinesisClient.describeStream(_streamName).getStreamDescription.getShards.asScala.toSeq
   }
 
   def splitShard(shardId: String): Unit = {
@@ -298,7 +298,7 @@ class SimpleDataGenerator(client: AmazonKinesisClient) extends KinesisDataGenera
       sentSeqNumbers += ((num, seqNumber))
     }
 
-    shardIdToSeqNumbers.toMap
+    shardIdToSeqNumbers.toMap.mapValues(_.toSeq).toMap
   }
 }
 
@@ -341,7 +341,8 @@ class AggregateDataGenerator( client: AmazonKinesisClient) extends KinesisDataGe
     val sentSeqNumbers = shardIdToSeqNumbers.getOrElseUpdate(shardId, new ArrayBuffer[(String, String)]())
     sentSeqNumbers += ((pk, seqNumber))
 
-    shardIdToSeqNumbers.toMap
+    shardIdToSeqNumbers.toMap.mapValues(_.toSeq).toMap
+
   }
 }
 

src/test/scala/org/apache/spark/sql/connector/kinesis/it/KinesisSourceCrossAccountItSuite.scala

@@ -98,7 +98,7 @@ abstract class KinesisSourceCrossAccountItSuite(
     }
 
     logInfo(s"Pushed data ${data.mkString("Array(", ", ", ")")}:\n\t${shardIdToSeqNumbers.mkString("\n\t")}")
-    shardIdToSeqNumbers.toMap
+    shardIdToSeqNumbers.toMap.mapValues(_.toSeq).toMap
   }
 
   def createCrossAccountSparkCustomReadStream(consumerType: String,

src/test/scala/org/apache/spark/sql/connector/kinesis/retrieval/EfoRecordBatchPublisherSuite.scala

@@ -339,7 +339,7 @@ class EfoRecordBatchPublisherSuite extends KinesisTestBase {
     events += createSubscribeToShardEvent(record)
     events += createSubscribeToShardEvent(record)
 
-    val kinesisClient = singleShardWithEvents(events)
+    val kinesisClient = singleShardWithEvents(events.toSeq)
     val now: Date = new Date
 
     val hashKeyRange: HashKeyRange = HashKeyRange.builder()
@@ -395,7 +395,7 @@ class EfoRecordBatchPublisherSuite extends KinesisTestBase {
     events += createSubscribeToShardEvent(record)
     events += createSubscribeToShardEvent(record)
 
-    val kinesisClient = singleShardWithEvents(events)
+    val kinesisClient = singleShardWithEvents(events.toSeq)
     val now: Date = new Date
 
     // Create ShardHandle with HashKeyRange excluding single UserRecord with hash key 0
@@ -755,7 +755,7 @@ class EfoRecordBatchPublisherSuite extends KinesisTestBase {
   }
 
   private def flattenToUserRecords(recordBatch: util.List[RecordBatch]): Seq[KinesisUserRecord] = {
-    recordBatch.asScala.flatMap(_.userRecords)
+    recordBatch.asScala.flatMap(_.userRecords).toSeq
   }
 
 

src/test/scala/org/apache/spark/sql/connector/kinesis/retrieval/client/FakeEfoClientConsumerFactory.scala

@@ -274,7 +274,7 @@ object FakeEfoClientConsumerFactory {
 
         batchIndex += 1
       }
-      events
+      events.toSeq
     }
   }
 

src/test/scala/org/apache/spark/sql/connector/kinesis/retrieval/client/FakePollingClientConsumerFactory.scala

@@ -157,7 +157,7 @@ object FakePollingClientConsumerFactory {
     }
 
     override def getShards: Seq[Shard] = {
-        dummyShards.map(_.shard)
+        dummyShards.map(_.shard).toSeq
       }
 
     override def getShardIterator(shardId: String,
@@ -180,7 +180,7 @@ object FakePollingClientConsumerFactory {
           .sequenceNumber(i.toString)
           .build
       }
-      batch
+      batch.toSeq
     }
   }
 
@@ -315,7 +315,7 @@ object FakePollingClientConsumerFactory {
                 .sequenceNumber(sequenceNumber.getAndAdd(numOfChildRecords).toString)
                 .build
       }
-      recordBatch
+      recordBatch.toSeq
     }
 
   }
@@ -418,7 +418,7 @@ object FakePollingClientConsumerFactory {
     }
 
     GetRecordsResponse.builder()
-      .records(records: _*)
+      .records(records.toSeq: _*)
       .millisBehindLatest(0L)
       .nextShardIterator(nextShardIterator)
       .build
@@ -427,7 +427,7 @@ object FakePollingClientConsumerFactory {
     override def getShards: Seq[Shard] = {
       listOfShards.map { streamShards =>
         streamShards.shard
-      }
+      }.toSeq
     }
   }
 }