Update fs2-core, fs2-io to 3.5.0

build.sbt

@@ -10,7 +10,7 @@ organization := "io.laserdisc"
 name         := "fs2-aws"
 
 lazy val scala213 = "2.13.10"
-lazy val scala3   = "3.1.1"
+lazy val scala3   = "3.2.1"
 
 lazy val supportedScalaVersions = List(scala213, scala3)
 
@@ -288,7 +288,7 @@ lazy val commonSettings = Def.settings(
   },
   Test / console / scalacOptions := (Compile / console / scalacOptions).value,
   Test / scalacOptions           := (Compile / scalacOptions).value,
-  Test / scalacOptions += "-Wconf:msg=is not declared `infix`:s,msg=is declared 'open':s",
+  Test / scalacOptions += "-Wconf:msg=is not declared infix:s,msg=is declared 'open':s",
   libraryDependencies += "org.scala-lang.modules" %% "scala-collection-compat" % "2.8.1",
   libraryDependencies ++= Seq(
     compilerPlugin(("org.typelevel" %% "kind-projector" % "0.13.2").cross(CrossVersion.full)),

fs2-aws-dynamodb/src/main/scala/fs2/aws/dynamodb/DynamoDB.scala

@@ -25,7 +25,6 @@ trait DynamoDB[F[_]] {
   /** Initialize a worker and start streaming records from a DynamoDB stream
     * On stream finish (due to error or other), worker will be shutdown
     *
-    * @tparam F effect type of the fs2 stream
     * @param appName    name of the DynamoDB application. Used by KCL when resharding
     * @param streamName name of the DynamoDB stream to consume from
     * @return an infinite fs2 Stream that emits Kinesis Records
@@ -35,8 +34,7 @@ trait DynamoDB[F[_]] {
   /** Initialize a worker and start streaming records from a DynamoDB stream
     * On stream finish (due to error or other), worker will be shutdown
     *
-    * @tparam F effect type of the fs2 stream
-    * @param consumerConfig configuration parameters for the KCL
+    * @param workerConfiguration configuration parameters for the KCL
     * @return an infinite fs2 Stream that emits DynamoDB Records
     */
   def readFromDynamoDBStream(
@@ -48,7 +46,6 @@ trait DynamoDB[F[_]] {
     * After accumulating maxBatchSize or reaching maxBatchWait for a respective shard, the latest record is checkpointed
     * By design, all records prior to the checkpointed record are also checkpointed in Kinesis
     *
-    * @tparam F effect type of the fs2 stream
     * @param checkpointSettings configure maxBatchSize and maxBatchWait time before triggering a checkpoint
     * @return a stream of Record types representing checkpointed messages
     */
@@ -79,7 +76,7 @@ object DynamoDB {
       // Instantiate a new bounded queue and concurrently run the queue populator
       // Expose the elements by dequeuing the internal buffer
       for {
-        dispatcher      <- Stream.resource(Dispatcher[F])
+        dispatcher      <- Stream.resource(Dispatcher.parallel[F])
         buffer          <- Stream.eval(Queue.unbounded[F, CommittableRecord])
         interruptSignal <- Stream.eval(SignallingRef[F, Boolean](false))
         _               <- instantiateScheduler(dispatcher, buffer, interruptSignal)

fs2-aws-dynamodb/src/main/scala/fs2/aws/dynamodb/StreamScan.scala

@@ -34,7 +34,7 @@ object StreamScan {
           pageSize: Int
       ): Stream[F, Chunk[JMap[String, AttributeValue]]] =
         for {
-          dispatcher <- Stream.resource(Dispatcher[F])
+          dispatcher <- Stream.resource(Dispatcher.parallel[F])
           queue      <- Stream.eval(Queue.bounded[F, Option[Chunk[JMap[String, AttributeValue]]]](1))
           sub        <- Stream.eval(Ref[F].of[Option[Subscription]](None))
           _ <- Stream.eval(

fs2-aws-dynamodb/src/test/scala/fs2/aws/dynamodb/NewDynamoDBConsumerSpec.scala

@@ -9,12 +9,7 @@ import com.amazonaws.services.dynamodbv2.streamsadapter.model.RecordAdapter
 import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorCheckpointer
 import com.amazonaws.services.kinesis.clientlibrary.interfaces.v2.{IRecordProcessor, IRecordProcessorFactory}
 import com.amazonaws.services.kinesis.clientlibrary.lib.worker.{ShutdownReason, Worker}
-import com.amazonaws.services.kinesis.clientlibrary.types.{
-  ExtendedSequenceNumber,
-  InitializationInput,
-  ProcessRecordsInput,
-  ShutdownInput
-}
+import com.amazonaws.services.kinesis.clientlibrary.types.{ExtendedSequenceNumber, InitializationInput, ProcessRecordsInput, ShutdownInput}
 import com.amazonaws.services.kinesis.model.Record
 import org.mockito.ArgumentMatchers.any
 import org.mockito.Mockito.*
@@ -27,7 +22,7 @@ import org.slf4j.LoggerFactory
 
 import java.util.Date
 import java.util.concurrent.{CountDownLatch, Phaser}
-import scala.concurrent.ExecutionContext
+import scala.concurrent.{ExecutionContext, Future}
 import scala.concurrent.duration.*
 import scala.jdk.CollectionConverters.*
 
@@ -118,7 +113,7 @@ class NewDynamoDBConsumerSpec
             recordProcessor.processRecords(recordsInput.withRecords(List(record).asJava))
           }
         }
-      ).parMapN { case (msgs, _, _) => msgs }.unsafeToFuture().futureValue
+      ).parMapN { case (msgs, _, _) => msgs }.unsafeToFuture().onError(e => Future(e.printStackTrace())).futureValue
 
     res should have size 60
   }

fs2-aws-kinesis/src/main/scala/fs2/aws/kinesis/Kinesis.scala

@@ -85,7 +85,7 @@ object Kinesis {
       // Instantiate a new bounded queue and concurrently run the queue populator
       // Expose the elements by dequeuing the internal buffer
       for {
-        dispatcher      <- Stream.resource(Dispatcher[F])
+        dispatcher      <- Stream.resource(Dispatcher.parallel[F])
         buffer          <- Stream.eval(Queue.bounded[F, Chunk[CommittableRecord]](streamConfig.bufferSize))
         interruptSignal <- Stream.eval(SignallingRef[F, Boolean](false))
         _               <- instantiateScheduler(dispatcher, buffer, interruptSignal)

project/Dependencies.scala

@@ -8,7 +8,7 @@ object Dependencies {
     val AwsSdk           = "2.18.7"
     val Circe            = "0.14.3"
     val Munit            = "0.7.29"
-    val Fs2              = "3.3.0"
+    val Fs2              = "3.5.0"
     val Refined          = "0.10.1"
     val ScalaTest        = "3.2.14"
     val MockitoScalaTest = "1.17.5"