如何解决KafkaConsumer.commitAsync行为的偏移量比以前低 如何复制如何避免使用commitAsync提交较低的偏移量
kafka将如何处理对的呼叫
KafkaConsumer.commitAsync(Map<TopicPartition,OffsetAndMetadata> offsets,OffsetCommitCallback callback)
给定主题的偏移量值小于上一次调用时的值?
解决方法
它将简单地将分区的偏移量设置为您指定的值,因此下次您将消耗来自commitedOffset + 1的消息。commitAsync()
的Javadoc说:
,提交的偏移量应该是应用程序将使用的下一条消息,即lastProcessedMessageOffset +1。
我很好奇,并对其进行了测试以查看其行为。正如文档中所写,@ haoyuwang在其答案(+1)中写的是正确的。
其背后的原因很简单。使用者组的已提交偏移量存储在内部主题__consumer_offsets
中的Kafka中。该主题为compact
,这意味着它旨在为给定密钥提供最新值。在您的情况下,键是使用者组,主题和分区的组合,而您的值是偏移量。
如果您现在
- 提交偏移量10,并且由于稍后进行异步处理
- 提交偏移量5
偏移量5将是__consumer_offsets
主题中的最新值。这意味着您的消费者将从该主题分区读取的下一个偏移量是偏移量6,而不是偏移量11。
如何复制
您可以简单地重现它并通过在常规提交之后(同步)提交更早的偏移量来对其进行测试,如下所示:
consumer.commitSync();
consumer.commitSync(commitFirstMessage);
其中commitFirstMessage
被定义为
TopicPartition zeroTopicPartition = new TopicPartition(topic,0);
OffsetAndMetadata zeroOffset = new OffsetAndMetadata(0L);
Map<TopicPartition,OffsetAndMetadata> commitFirstMessage = new HashMap<>();
commitFirstMessage.put(zeroTopicPartition,zeroOffset);
编辑:
如何避免使用commitAsync提交较低的偏移量
在书Kafka - The Definitive Guide中,建议避免由于重试调用commitAsync
而提交较低的偏移量:
重试异步提交:获得异步重试正确的提交顺序的一种简单模式是使用单调递增的序列号。每次提交时增加序列号,并添加 提交到commitAsync回调时的序列号。当您准备发送重试时,请检查回调获得的提交序列号是否与实例变量相等;如果是,则没有较新的提交,可以重试。如果实例序列号更高,请不要重试,因为已经发送了新的提交。
一个实现可能看起来像这样(未经实际测试!):
import java.util._
import java.time.Duration
import org.apache.kafka.clients.consumer.{ConsumerConfig,ConsumerRecord,KafkaConsumer,OffsetAndMetadata,OffsetCommitCallback}
import org.apache.kafka.common.{KafkaException,TopicPartition}
import collection.JavaConverters._
object AsyncCommitWithCallback extends App {
// define topic
val topic = "myOutputTopic"
// set properties
val props = new Properties()
props.put(ConsumerConfig.GROUP_ID_CONFIG,"AsyncCommitter5")
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092")
// [set more properties...]
// create KafkaConsumer and subscribe
val consumer = new KafkaConsumer[String,String](props)
consumer.subscribe(List(topic).asJavaCollection)
// initialize global counter
val atomicLong = new AtomicLong(0)
// consume message
try {
while(true) {
val records = consumer.poll(Duration.ofMillis(1)).asScala
if(records.nonEmpty) {
for (data <- records) {
// do something with the records
}
consumer.commitAsync(new KeepOrderAsyncCommit)
}
}
} catch {
case ex: KafkaException => ex.printStackTrace()
} finally {
consumer.commitSync()
consumer.close()
}
class KeepOrderAsyncCommit extends OffsetCommitCallback {
// keeping position of this callback instance
val position = atomicLong.incrementAndGet()
override def onComplete(offsets: util.Map[TopicPartition,OffsetAndMetadata],exception: Exception): Unit = {
// retrying only if no other commit incremented the global counter
if(exception != null){
if(position == atomicLong.get) {
consumer.commitAsync(this)
}
}
}
}
}
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。