block22.scala

// ... previous code
val consumerM = Consumer.make(consumerSettings)

for {
  _ <- session.execute(Sql.partitionTableDdl)
  _ <- session.execute(Sql.balanceTableDdl)

  _ <- consumerM.use { consumer =>
    consumer.subscribeAnd(Subscription.topics(TopicName))
      .plainStream(Serde.int, CommandSerde)
      .tap(x => log.info(x.value.toString))
      .runDrain
  }
} yield ()