为什么自动提交启用了 Kafka 客户端在消费者关闭期间提交最新生成的消息的偏移量，即使消息尚未被使用？

Question

TLDR:

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

详细解释：

我有一个简单的scala应用程序，它有一个Akka actor，它使用来自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秒，然后停下来。

Starter.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
}
}

具有AutoCommit.scala 的BaseActor

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的消息，但在下一个周期中没有消耗

事实上，我不是阿卡演员的专家，也不知道这种情况是来自卡夫卡还是阿卡，但似乎与自动提交给我有关。

---

使用的依赖版本：

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,
)
}

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

Answer 1

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

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

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

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

Alpakka-Kafka文档对至少一次处理进行了深入的讨论：在这种情况下，至少一次的处理可能需要引入手动偏移提交并将mapAsyncUnordered替换为mapAsync(因为mapAsyncUnordered与手动偏移提交结合意味着您的应用程序只能保证来自Kafka的消息至少被处理零次(。

在Alpakka-Kafka中，有一个广泛的消息处理保证分类法：

• 最多硬一次：Consumer.atMostOnceSource-处理前在每条消息后提交
• 最多软一次：enable.auto.commit = true-"；柔软的"；因为提交实际上是为了增加吞吐量而批处理的，所以这实际上是"；最多一次，除非至少一次
• 至少硬一次：只有在所有处理都验证成功后才手动提交
• soft至少一次：在一些处理完成后手动提交(即"至少一次，除非最多一次"(
• 正好一次：一般情况下是不可能的，但如果您的处理能够重复数据消除，从而使重复数据成为幂等的，则可以有效地进行一次