KafkaStreams EXACTLY_ONCE 保证 - 跳过 kafka 偏移量

Question

我正在使用 Spark 2.2.0 和 kafka 0.10 spark-streaming 库来读取充满 Kafka-Streams scala 应用程序的主题。 Kafka Broker 版本为 0.11，Kafka-streams 版本为 0.11.0.2。

当我在 Kafka-Stream 应用程序中设置 EXACTLY_ONCE 保证时：

 p.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE)

我在 Spark 中收到此错误：

java.lang.AssertionError: assertion failed: Got wrong record for spark-executor-<group.id> <topic> 0 even after seeking to offset 24
at scala.Predef$.assert(Predef.scala:170)
at org.apache.spark.streaming.kafka010.CachedKafkaConsumer.get(CachedKafkaConsumer.scala:85)
at org.apache.spark.streaming.kafka010.KafkaRDD$KafkaRDDIterator.next(KafkaRDD.scala:223)
at org.apache.spark.streaming.kafka010.KafkaRDD$KafkaRDDIterator.next(KafkaRDD.scala:189)
at scala.collection.Iterator$class.foreach(Iterator.scala:893)
at org.apache.spark.streaming.kafka010.KafkaRDD$KafkaRDDIterator.foreach(KafkaRDD.scala:189)
at scala.collection.generic.Growable$class.$plus$plus$eq(Growable.scala:59)
at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:104)
at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:48)
at scala.collection.TraversableOnce$class.to(TraversableOnce.scala:310)
at org.apache.spark.streaming.kafka010.KafkaRDD$KafkaRDDIterator.to(KafkaRDD.scala:189)
at scala.collection.TraversableOnce$class.toBuffer(TraversableOnce.scala:302)
at org.apache.spark.streaming.kafka010.KafkaRDD$KafkaRDDIterator.toBuffer(KafkaRDD.scala:189)
at scala.collection.TraversableOnce$class.toArray(TraversableOnce.scala:289)
at org.apache.spark.streaming.kafka010.KafkaRDD$KafkaRDDIterator.toArray(KafkaRDD.scala:189)
at org.apache.spark.rdd.RDD$$anonfun$collect$$anonfun.apply(RDD.scala:936)
at org.apache.spark.rdd.RDD$$anonfun$collect$$anonfun.apply(RDD.scala:936)
at org.apache.spark.SparkContext$$anonfun$runJob.apply(SparkContext.scala:2062)
at org.apache.spark.SparkContext$$anonfun$runJob.apply(SparkContext.scala:2062)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
at org.apache.spark.scheduler.Task.run(Task.scala:108)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:335)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)

如果没有设置EXACTLY_ONCE 属性，它工作得很好。

编辑 1：充满 kafka-streams 应用程序的主题（恰好启用一次）有错误的结束偏移量。当 i 运行 kafka.tools.GetOffsetShell 时，它给出结束偏移量 18，但在主题中只有 12 条消息（保留被禁用）。当 exactly once 保证被禁用时，这些偏移量是匹配的。我尝试根据this重置kafka-streams，但问题仍然存在。

编辑 2：当我运行 SimpleConsumerShell 使用 --print-offsets 选项时，输出如下：

next offset = 1
{"timestamp": 149583551238149, "data": {...}}
next offset = 2
{"timestamp": 149583551238149, "data": {...}}
next offset = 4
{"timestamp": 149583551238149, "data": {...}}
next offset = 5
{"timestamp": 149583551238149, "data": {...}}
next offset = 7
{"timestamp": 149583551238149, "data": {...}}
next offset = 8
{"timestamp": 149583551238149, "data": {...}}
...

启用精确一次交付保证时，显然会跳过一些偏移量。

有什么想法吗？什么会导致这个？谢谢！

Answer 1

我发现偏移间隙是 Kafka（版本 >= 0.11）中的预期行为，这些是由 commit/abort 事务标记引起的。

有关 kafka 事务和控制消息的更多信息here：

These transaction markers are not exposed to applications, but are used by consumers in read_committed mode to filter out messages from aborted transactions and to not return messages which are part of open transactions (i.e., those which are in the log but don’t have a transaction marker associated with them).

和here.

Kafka 事务是在 Kafka 0.11 中引入的，所以我假设 spark-streaming-kafka 库 0.10 不兼容此消息格式，并且尚未实现较新版本的 spark-streaming-kafka。

KafkaStreams EXACTLY_ONCE 保证 - 跳过 kafka 偏移量

KafkaStreams EXACTLY_ONCE guarantee - skipping kafka offsets

offset

apache-kafka

spark-streaming

apache-kafka-streams