Как безопасно пропускать сообщения с помощью Lagom Kafka Message Broker API?

Мы определили базового подписчика, который пропускает ошибочные сообщения (т. е. по какой-то причине бизнес-логики, которую мы не собираемся обрабатывать), выбрасывая исключение и полагаясь на контроль потока Akka Streams для возобновления Flow:

someLagomService
  .someTopic()
  .subscribe
  .withGroupId("lagom-service")
  .atLeastOnce(
    Flow[Int]
      .mapAsync(1)(el => {
        // Exception may occur here or can map to Done
      })
      .withAttributes(ActorAttributes.supervisionStrategy({
        case t =>
          Supervision.Resume
      })
  )

Кажется, это хорошо работает для основных случаев использования при очень небольшой нагрузке, но мы заметили очень странные вещи для большого количества сообщений (например, очень частая повторная обработка сообщений и т. д.).

Покопавшись в коде, мы увидели, что в документации Lagom broker.Subscriber.atLeastOnce говорится:

The flow may pull more elements from upstream but it must emit exactly one Done message for each message that it receives. It must also emit them in the same order that the messages were received. This means that the flow must not filter or collect a subset of the messages, instead it must split the messages into separate streams and map those that would have been dropped to Done.

Кроме того, в реализации Lagom KafkaSubscriberActor мы видим, что реализация private atLeastOnce по существу распаковывает полезную нагрузку и смещение сообщения, а затем снова архивирует и создает резервную копию после того, как наш пользовательский поток сопоставляет сообщения с Done.

Эти два лакомых кусочка выше, похоже, подразумевают, что, используя супервизоры потоков и пропуская элементы, мы можем оказаться в ситуации, когда коммитируемые смещения больше не будут синхронизироваться равномерно с Done, которые должны быть созданы для каждого сообщения Kafka.

Пример: если мы транслируем 1, 2, 3, 4 и сопоставляем 1, 2 и 4 с Done, но выбрасываем исключение для 3, у нас есть 3 Done и 4 фиксируемых смещения?

• Это правильно/ожидается? Означает ли это, что мы должны ИЗБЕГАТЬ использования здесь супервизоров потоков?
• Какое поведение может быть вызвано неравномерным сжатием?
• Каков рекомендуемый подход к обработке ошибок, когда речь идет об использовании сообщений из Kafka через API брокера сообщений Lagom? Правильно ли сопоставлять / восстанавливать сбои в Done?

Использование Лагом 1.4.10