apache-storm - Storm-kafka 0.8 plus，我可以从最新的偏移量读取吗？

Question

我有一个 Kafka spout 的拓扑结构，如下所示

SpoutConfig spoutConfig = new SpoutConfig(zkBrokerHosts, "some-topic","", "some-id");
spoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
...

builder.setSpout("kafkaSpout",new KafkaSpout(spoutConfig),1);

当然它工作正常。

考虑到我的拓扑失败并再次运行它的情况，我希望 KafkaSpout 从该主题的最新偏移量读取，而不是从消费者读取的最后一个偏移量读取。

有什么办法吗？我试过了

spoutConfig.startOffsetTime=System.currentTimeMillis();

但似乎并没有如我所愿。也不是 kafkaConfig.forceStartOffsetTime(-2);

如果你有什么想法，请告诉我。

Answer 1

尝试 kafkaConfig.forceStartOffsetTime(-1)。 -1 表示最新的 Kafka 偏移量，-2 表示最早的可用偏移量。

编辑:

此外，您可以使用相同的选项强制 spout 从任何所需的偏移量开始消费——只需将数字偏移量作为唯一参数传递即可。

忽略forceStartOffsetTime中的“Time”，参数名有点乱。 Kafka 中的偏移量是数字，与任何时间概念无关。 -1 只是告诉 Kafka spout 从 Kafka 本身收集最新偏移量的一种特殊方式(idem -2 用于最早的可用偏移量)。