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

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

trident offset apache-storm apache-zookeeper apache-kafka

phaigeim · 技术社区 · 6 年前

我正在使用storm kafka client 1.2.1,并为KafkaTridentSpoutOpaque创建喷口配置,如下所示

            kafkaSpoutConfig = KafkaSpoutConfig.builder(brokerURL, kafkaTopic)
                .setProp(ConsumerConfig.GROUP_ID_CONFIG,"storm-kafka-group")
                .setProcessingGuarantee(ProcessingGuarantee.AT_MOST_ONCE)
                .setProp(ConsumerConfig.CLIENT_ID_CONFIG,InetAddress.getLocalHost().getHostName())

我在卡夫卡和Zookeeper中都找不到我的组id和偏移量。通过Zookeeper,我尝试了zkCli。sh和尝试 ls /consumers 但并没有,因为我认为卡夫卡自己现在正在维持偏移量,而不是动物园管理员。

我也用下面的命令尝试过卡夫卡

bin/kafka-run-class.sh kafka.admin.ConsumerGroupCommand  --list  --bootstrap-server localhost:9092
Note: This will not show information about old Zookeeper-based consumers.
console-consumer-20130
console-consumer-82696
console-consumer-6106
console-consumer-67393
console-consumer-14333
console-consumer-21174
console-consumer-64550

有人能帮我找到偏移量吗?如果我重新启动拓扑,它会再次重播卡夫卡中的事件吗?

1 回复 | 直到 6 年前

Stig Rohde Døssing 6 年前

三叉戟在卡夫卡中不存储偏移量,但在斯托姆的动物园管理员中存储偏移量。如果使用Storm的Zookeeper配置的默认设置运行,则Storm的Zookeeper中的路径如下 /coordinator/<your-topology-id>/meta .

该路径下的对象将包含第一个和最后一个偏移量,以及每个批的主题分区。因此,例如。 /coordinator/<your-topology-id>/meta/15 将包含批次号15中发出的第一个和最后一个偏移量。

重新启动后喷口是否重放偏移量由 FirstPollOffsetStrategy 您在 KafkaSpoutConfig 。默认值为 UNCOMMITTED_EARLIEST ,重新启动时不会重新启动。查看Javadoc https://github.com/apache/storm/blob/v1.2.1/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpoutConfig.java#L126 .

推荐文章

Ajeesh · Apache Storm(1.2.1)运输异常断管

7 年前

TechCrap · apache storm与python混合拓扑-ModuleNotFoundError:没有名为“storm”的模块

7 年前

Aniruddha · 在storm群集上提交拓扑时出错

7 年前

kingluo · storm:bolt如何执行元组?

7 年前

Ahmad Osama · 将数据从Apache Storm插入Azure Cosmos DB

7 年前

Saurabh · 在Apache Storm中使用与多个螺栓相同的类

7 年前

Keshore Durairaj · 实时流数据聚合

7 年前

Siva S · Apache strom—包的返回类型。暴风雨元组不存在

7 年前

Ryanqy · 监管在风暴中意味着什么?

7 年前

jdowdell · 流处理架构:未来事件影响过去结果

7 年前