1

TLDR：

  • 对于启用自动提交的 Kafka 客户端，提交生成的消息的偏移量是否已消耗（即使不是）预期行为？（对于消费和生产相同主题的应用程序）

详细解释：

我有一个简单的 scala 应用程序，它有一个 Akka 演员，它使用来自 Kafka 主题的消息，如果在消息处理期间发生任何异常，则将消息生成到同一主题。

TestActor.scala

  override protected def processMessage(messages: Seq[ConsumerRecord[String, String]]): Future[Done] = {
    Future.sequence(messages.map(message => {
      logger.info(s"--CONSUMED: offset: ${message.offset()} message: ${message.value()}")
      // in actual implementation, some process is done here and if an exception occurs, the message is sent to the same topic as seen below
      sendToExceptionTopic(Instant.now().toEpochMilli)
      Thread.sleep(1000)
      Future(Done)
    })).transformWith(_ => Future(Done))
  }

该演员每分钟开始并运行 20 秒然后停止。

启动器.scala

  def init(): Unit = {
    exceptionManagerActor ! InitExceptionActors

    system.scheduler.schedule(2.second, 60.seconds) {
      logger.info("started consuming messages")
      exceptionManagerActor ! ConsumeExceptions
    }
  }

ExceptionManagerActor.scala

  private def startScheduledActor(actorRef: ActorRef): Unit = {
    actorRef ! Start

    context.system.scheduler.scheduleOnce(20.seconds) {
      logger.info("stopping consuming messages")
      actorRef ! Stop
    }
  }

BaseActorWithAutoCommit.scala

  override def receive: Receive = {
    case Start =>
      consumerBase = consumer
        .groupedWithin(20, 2000.millisecond)
        .mapAsyncUnordered(10)(processMessage)
        .toMat(Sink.seq)(DrainingControl.apply)
        .run()

    case Stop =>
      consumerBase.drainAndShutdown().transformWith {
        case Success(value) =>
          logger.info("actor stopped")
          Future(value)
        case Failure(ex) =>
          logger.error("error: ", ex)
          Future.failed(ex)
      }
    //Await.result(consumerBase.drainAndShutdown(), 1.minute)
  }

使用此配置，在停止时，Kafka 客户端将提交最新生成的消息的偏移量，就好像它已被消费一样。

示例日志：

14:28:48.868 INFO - started consuming messages
14:28:50.945 INFO - --CONSUMED: offset: 97 message: 1
14:28:51.028 INFO - ----PRODUCED: offset: 98 message: 1643542130945
...
14:29:08.886 INFO - stopping consuming messages
14:29:08.891 INFO - --CONSUMED: offset: 106 message: 1643542147106
14:29:08.895 INFO - ----PRODUCED: offset: 107 message: 1643542148891 <------ this message was lost
14:29:39.946 INFO - actor stopped
14:29:39.956 INFO - Message [akka.kafka.internal.KafkaConsumerActor$Internal$StopFromStage] from Actor[akka://test-consumer/system/Materializers/StreamSupervisor-2/$$a#1541548736] to Actor[akka://test-consumer/system/kafka-consumer-1#914599016] was not delivered. [1] dead letters encountered. If this is not an expected behavior then Actor[akka://test-consumer/system/kafka-consumer-1#914599016] may have terminated unexpectedly. This logging can be turned off or adjusted with configuration settings 'akka.log-dead-letters' and 'akka.log-dead-letters-during-shutdown'.
14:29:48.866 INFO - started consuming messages <----- The message with offset 107 was expected in this cycle to consume but it was not consumed
14:30:08.871 INFO - stopping consuming messages
14:30:38.896 INFO - actor stopped

从日志中可以看到，产生了一条偏移量为 107 的消息，但在下一个周期中没有被消费。

实际上，我不是 Akka 演员的专家，不知道这种情况是来自 Kafka 还是 Akka，但似乎与我的自动提交有关。

使用的依赖版本：

lazy val versions = new {
  val akka = "2.6.13"
  val akkaHttp = "10.1.9"
  val alpAkka = "2.0.7"
  val logback = "1.2.3"
  val apacheCommons = "1.7"
  val json4s = "3.6.7"
}

libraryDependencies ++= {
  Seq(
    "com.typesafe.akka" %% "akka-slf4j" % versions.akka,
    "com.typesafe.akka" %% "akka-stream-kafka" % versions.alpAkka,
    "com.typesafe.akka" %% "akka-http" % versions.akkaHttp,
    "com.typesafe.akka" %% "akka-protobuf" % versions.akka,
    "com.typesafe.akka" %% "akka-stream" % versions.akka,
    "ch.qos.logback" % "logback-classic" % versions.logback,
    "org.json4s" %% "json4s-jackson" % versions.json4s,
    "org.apache.commons" % "commons-text" % versions.apacheCommons,
  )
}

可以从此存储库获得示例源代码和重现该情况的步骤

4

1 回答 1

3

就 Kafka 而言，只要 Alpakka Kafka 从 Kafka 读取消息，消息就会被消耗掉。

这是在 Alpakka Kafka 内部的参与者将其发送给下游消费者以进行应用程序级处理之前。

因此， Kafka 自动提交 ( enable.auto.commit = true) 将导致在消息发送到您的参与者之前提交偏移量。

关于偏移量管理的 Kafka 文档确实（在撰写本文时）提到enable.auto.commit具有至少一次语义，但正如我在第一段中所指出的，这是一种至少一次交付语义，而不是至少一次 -一次处理语义。后者是应用程序级别的问题，要实现这一点需要延迟偏移提交，直到处理完成。

Alpakka Kafka 文档有一个关于至少一次处理的相关讨论：在这种情况下，至少一次处理可能需要引入手动偏移提交和替换mapAsyncUnordered为mapAsync（因为mapAsyncUnordered与手动偏移提交一起意味着您的应用程序只能保证来自 Kafka 的消息至少被处理零次）。

在 Alpakka Kafka 中，广泛的消息处理分类保证：

  • hard at-most-once: Consumer.atMostOnceSource- 在处理之前在每条消息之后提交
  • soft at-most-once：enable.auto.commit = true-“soft”，因为提交实际上是为提高吞吐量而进行批处理的，所以这实际上是“at-most-once，除非它是 at-least-once”
  • hard at-least-once：只有在所有处理都被验证成功后才手动提交
  • soft at-least-once：在某些处理完成后手动提交（即“至少一次，除非它是最多一次”）
  • 完全一次：通常不可能，但如果您的处理有重复数据删除的方法，从而使重复项具有幂等性，您可以有效地一次
于 2022-01-31T17:58:46.283 回答