ConsumerApplication.scala

import cats.effect.{ConcurrentEffect, ExitCode, IO, IOApp}
import scala.concurrent.{ExecutionContext, ExecutionContextExecutor}

object ConsumerApplication extends IOApp {

  implicit private val ec: ExecutionContextExecutor = ExecutionContext.global
  private val cs = IO.contextShift(ec)
  implicit private val concurrentEffect: ConcurrentEffect[IO] = IO.ioConcurrentEffect(cs)

  val config = ConsumerConfig("localhost:9092", "test-topic", "consumer-group-1")

  override def run(args: List[String]): IO[ExitCode] = {
    val applicationResource = for {
      kafkaContext <- KafkaContext.resource(cs)
      consumer <- Consumer.resource(new MessageHandler, config, kafkaContext)
    } yield consumer
    applicationResource.use(consumer => consumer.subscribe *> consumer.consume).as(ExitCode.Success)
  }

}