代码之家 › 专栏 › 技术社区 › JJ15k

卡夫卡分区分配协议

kafka-producer-api kafka-consumer-api apache-kafka c#

JJ15k · 技术社区 · 6 年前

我正在编写集成测试,以使用confluent dotnet(包装librdkafka)验证卡夫卡生产者-消费者配置。

在一个测试中,我想启动一个消费者,该消费者将从现有分区的末尾开始,然后发布一条来自生产者的消息,并断言消费者只使用了一条消息。

现在,使用者的启动是异步的(即:如果您调用subscribe然后立即发布,则从末尾开始的使用者将不会收到它)。在没有竞争条件的情况下,编写此测试的适当方法是什么?完成“分区.分配”后,是否已确定使用者偏移量?我不确定,因为OnPartitionAssigned的回调只包含TopicPartition,没有偏移量。

在一个相关的问题上,似乎有时,在没有任何kafka节点故障的情况下(afaict),我收到的分配的分区比分配的分区多(即:我两次分配相同的分区),这怎么可能?

1 回复 | 直到 6 年前

Edenhill 6 年前

设置OnPartitionEof委托,当使用者到达分区的末尾时将调用该委托,当调用该委托时,您可以确保使用者确实正在获取给定分区的消息,并且可以开始向其生成消息。

        consumer.OnPartitionEOF += (_, end)
            => Console.WriteLine($"Reached end of topic {end.Topic} partition {end.Partition}, next message will be at offset {end.Offset}");

推荐文章

yershalom · 同一消费群体中具有不同分区数量的不同卡夫卡主题

2 年前

Oskars Apinis · 将kafka从服务器迁移到docker容器

2 年前

Sravan Kumar · Apache kafka模式注册表问题

2 年前

Panadol Chong · 无法建立到节点1(localhost/127.0.0.1:9092)的连接。经纪人可能不在。甚至传入正确的端点[重复]

2 年前

MrSham · 如何确保旧版本0.10.0在卡夫卡中只处理一次

2 年前

TheTags · Kafka-PLC4X-Modbus/TCP-作业刮取期间异常

2 年前

Iheb Mar · 卡夫卡制作人/消费者粉碎每一秒的API调用

2 年前

hasherBaba · 在为Nifi编写流媒体服务时,如何解决Java的空指针异常?[闭门]

6 年前

phaigeim · 为storm consumer检查卡夫卡主题的偏移量

6 年前

Raman · Kafka流:POJO序列化/反序列化

6 年前