block21.scala

def partitionToString(topic: TopicPartition): String =
  s"${topic.topic()}-${topic.partition()}"

override def run(args: List[String]): URIO[zio.ZEnv, ExitCode] = {
  val zio = for {
    session <- ZIO.service[SessionTask]

    _ <- ZIO.runtime[Any].flatMap { implicit r: Runtime[Any] =>
      val manualOffsetRetrieval = Consumer.OffsetRetrieval.Manual { partitions =>
        val list = partitions.map(partitionToString).toList
        val query = Sql.selectPartitionSql(list)
        session
          .prepare(query).toManagedZIO
          .use(_.stream(list, 64).compile.toVector)
          .map(xs => xs.collect { case Some(t) => t }
            .map(x => x.topic -> x.offset)
            .toMap)
      }
      val consumerSettings = ConsumerSettings(List("localhost:9092"))
        .withGroupId("my.group")
        .withOffsetRetrieval(manualOffsetRetrieval)
      val consumerM = Consumer.make(consumerSettings)

      ??? // to be continued
    }
  } yield ()

  ??? // to be continued
}