lagom/lagom

Hello. I have a need to consume from a kafka topic (still within same cluster, but not published via events).

Documentations suggest that I create an ExternalKafkaService with the topic defined in its descriptor, then I can write my subscriber as usual. This works, but I'm just wondering if we have any way to not depending on Lagom for this purpose.

  logger.info("Starting the subscriber")
  externalKafkaService.externalTopic.subscribe.atLeastOnce(Flow[ExternalRequest].mapAsync(1) {
    request =>
      logger.info("Executing {}", request)
      Future.successful(Done)
  })

If I implement the subscriber like this, I can see the "starting" log on startup, but not the "executing" log when I produce the messages to the topic (works fine if using Lagom way)

  private val consumerSettings = ConsumerSettings(actorSystem, new StringDeserializer, new StringDeserializer)

  logger.info("Starting the subscriber")
  Consumer
    .plainSource(consumerSettings, Subscriptions.topics(ExternalTopic))
    .mapAsync(1)(message => {
      val request = Json.parse(message.value).as[ExternalRequest]
      logger.info("Executing {}", request)
      Future.successful(Done)
    })
    .run()

Could you help me understand what is missing in 2nd approach? Is there any preferences/advantages to use Lagom way instead, even if we can make 2nd approach work? (I can change 2nd approach to use commitableSource to achieve the atLeastOnce behaviour later)
Thank you in advance.