dongliming2416 2017-02-01 15:46
浏览 736
已采纳

如何在Golang Kafka 10中将使用者设置为从特定偏移量开始

My need is to make the producer to start from the last message it processed before it crashed. Fortunately I am in the case of having only one topic, with one partition and one consumer.

To do so I tried https://github.com/Shopify/sarama but it doesn't seems to be available yet. I am now using https://godoc.org/github.com/bsm/sarama-cluster, which allow me to commit every message offset.

I cannot retrieve the last committed offset I cannot figure out how to make a sarama consumer to start from said offset. The only parameter I've found so far is Config.Producer.Offsets.Initial.

  1. How to retrieve the last committed offset?
  2. How to make the consumer start from the last message whose offset has been committed? OffsetNewest will make it start from the last message produced, not the last processed b the consumer.
  3. Is it possible to do so using only Shopify/sarama and not bsm/sarama-cluster ?

Thank in advance

P.S. I am using Kafka 10.0, so the offsets are stores in a kafka and not in zookeeper.

EDIT1: Partial solution: fetch all the messages since sarama.OffsetOldest and skip all of them until we found a non processed one.

  • 写回答

1条回答 默认 最新

  • dowb58485 2018-01-18 13:27
    关注

    If offset was already saved for a partition, sarama-cluster will resume consumption from that offset. The Config.Producer.Offsets.Initial option is used only if no saved offset is present (first run for a consumer group).

    You can verify this by adding the following line at the beginning of your main() function:

    sarama.Logger = log.New(os.Stdout, "sarama: ", log.LstdFlags)
    

    Then you'll see something like the following in the output:

    cluster/consumer CID-17db1be4-a162-411c-a106-4d198191176a consume sample/0 from 12

    The 12 in that is the offset Sarama is going to start from for that partition (sample/0).

    本回答被题主选为最佳回答 , 对您是否有帮助呢?
    评论

报告相同问题?

悬赏问题

  • ¥15 数值计算离散正交多项式
  • ¥30 数值计算均差系数编程
  • ¥15 redis-full-check比较 两个集群的数据出错
  • ¥15 Matlab编程问题
  • ¥15 训练的多模态特征融合模型准确度很低怎么办
  • ¥15 kylin启动报错log4j类冲突
  • ¥15 超声波模块测距控制点灯,灯的闪烁很不稳定,经过调试发现测的距离偏大
  • ¥15 import arcpy出现importing _arcgisscripting 找不到相关程序
  • ¥15 onvif+openssl,vs2022编译openssl64
  • ¥15 iOS 自定义输入法-第三方输入法