KafkaStorm喷口改变拓扑和消耗从旧的偏移

Question

我用Kafka壶嘴来消费信息。但是，如果我必须更改拓扑并上传，那么它将从旧消息恢复还是从新消息开始？Kafka壶嘴给了我们从哪里消费的时间戳，但我怎么知道时间戳呢？

Answer 1

基本上，事件的顺序将是：

>

第一次从以下属性开始读取拓扑：

forceFromStart = true

startOffsetTime = -2

以上道具将迫使它从话题的开头开始。请记住具有这两个属性，因为forceFromStart告诉Storm读取starOffsetTime属性，并使用设置的值来确定从哪里开始读取，并忽略zoowatch偏移。

从现在起，拓扑将运行，zookeeper将保持偏移。如果您的工作人员死亡，它将由主管启动，并从zookeeper中的偏移量开始读取。

现在，如果要重新启动拓扑，并且要从关机前停止的位置读取拓扑，请使用以下属性并重新启动拓扑：

forceFromStart = false

通过上面的属性，您告诉Storm不要读取starOffsetTime值，而是使用在关闭拓扑结构之前维护的zoowatch偏移量。

从现在开始，每次重新启动拓扑，它都将从它离开的地方读取。

如果要重新启动拓扑，并且要从主题的开头/顶部读取，请使用以下属性并重新启动拓扑：

forceFromStart = true

startOffsetTime = -1

通过上面的属性，你告诉Storm忽略动物园管理员的偏移量，从主题的最新偏移量开始。

Answer 2

如果您使用KafkaSpout，请确保以下内容：

1. 在SpoutConfig中，"id"和"zkroot"在重新部署新版本的拓扑后不会更改。Storm使用zkroot、id将主题偏移量存储到zoowatch中
2. KafkaConfig.forceFromStart设置为false。

KafkaSpout将偏移量存储到动物园管理员。在重新部署期间要非常小心，如果您在KafkaSpout的KafkaConfig中将forceFromStart设置为true（这可能是您第一次部署拓扑时的情况），它将忽略存储的动物园管理员偏移。确保设置为false。

考虑写你的拓扑结构，让KAFKACONFIG。当拓扑的main（）方法执行时，将从属性文件中读取forceFromStart值。这将允许管理员控制是否重播Kafka消息。

Answer 3

spoutConfig。forceStartOffsetTime（-1）

它将选择围绕该时间戳写入的最新偏移量来开始消费。您可以通过传入-1来强制喷口总是从最近的偏移开始，也可以通过传入-2来强制喷口从最早的偏移开始。

参考文献