Update Cats Effect (3.5.x) and FS2

[calvinlfer]

• Update Cats Effect to 3.5.3
• Update FS2 to 3.9.4 and use updated APIs

Memory usage continues to be stable with the update to the latest Cats Effect 3.5.x version

libraryDependencies ++= 
  Seq(
    "dev.zio"         %% "zio"               % "2.0.21",
    "dev.zio"         %% "zio-interop-cats"  % "23.1.0.1+0-660c3b77+20240218-1931-SNAPSHOT",
    "dev.zio"         %% "zio-logging-slf4j" % "2.2.0",
    "com.github.fd4s" %% "fs2-kafka"         % "3.2.0",
    "ch.qos.logback"   % "logback-classic"   % "1.4.14"
  )

import fs2.Stream
import fs2.kafka.*
import zio.{durationInt as _, *}
import zio.interop.catz.*
import zio.logging.backend.SLF4J

import scala.concurrent.duration.*

object ReproducerFs2KafkaZIO extends ZIOAppDefault:
  override val bootstrap: ZLayer[ZIOAppArgs, Any, Any] =
    Runtime.removeDefaultLoggers ++ SLF4J.slf4j

  val topic = "example-topic"

  val consumer: Stream[Task, Unit] =
    val settings =
      ConsumerSettings[Task, String, String]
        .withBootstrapServers("localhost:9092")
        .withGroupId("test-consumer-group-id-2")
        .withAutoOffsetReset(AutoOffsetReset.Earliest)

    KafkaConsumer
      .stream[Task, String, String](settings)
      .evalTap(_.subscribeTo(topic))
      .stream
      .mapChunks(_.map(_.offset))
      .through(commitBatchWithin[Task](2048, 10.seconds))

  val producer: Stream[Task, ProducerResult[String, String]] =
    val settings =
      ProducerSettings[Task, String, String]
        .withBootstrapServers("localhost:9092")
        .withBatchSize(128)
        .withAcks(Acks.One)
        .withEnableIdempotence(false)
        .withRetries(128)

    val producerPipe = KafkaProducer.pipe[Task, String, String](settings)

    Stream
      .iterate[Task, Long](0L)(_ + 1L)
      .map(n => ProducerRecord(topic = topic, key = s"key: $n", value = s"value: $n"))
      .chunkN(n = 128, allowFewer = true)
      .map(ProducerRecords[fs2.Chunk, String, String])
      .through(producerPipe)

  override val run: Task[Unit] =
    Stream(producer, consumer).parJoinUnbounded.compile.drain