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 本身收集最新偏移量的一种特殊方式（同上 -2 表示最早可用的偏移量）。