为什么我的Kafka留言都在暴风里重播？

Question

我试图弄清楚为什么每次我重新启动我的Storm拓扑时，我所有的Kafka消息都被重播。

我的理解是，一旦最后一个Bolt对元组进行了处理，spout就应该在Kafka上提交消息，因此在重新启动后我不应该看到它的重播。

我的代码是一个简单的Kafka喷口和一个螺栓，只是打印每一个消息，然后处理他们。

private static KafkaSpout buildKafkaSpout(String topicName) {
    ZkHosts zkHosts = new ZkHosts("localhost:2181");
    SpoutConfig spoutConfig = new SpoutConfig(zkHosts, 
            topicName, 
            "/" + topicName, 
            "mykafkaspout");      /*was:UUID.randomUUID().toString()*/
    spoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
    return new KafkaSpout(spoutConfig);
}

public static class PrintBolt extends BaseRichBolt {
    OutputCollector _collector;
    public static Logger LOG = LoggerFactory.getLogger(PrintBolt.class);

    @Override
    public void prepare(Map conf, TopologyContext context, OutputCollector collector) {
        _collector = collector;
    }

    @Override
    public void execute(Tuple tuple) {
        LOG.error("PrintBolt.0: {}",tuple.getString(0));
        _collector.ack(tuple);
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("nothing"));
    }
}

public static void main(String[] args) throws Exception {
    TopologyBuilder builder = new TopologyBuilder();

    builder.setSpout("kafka", buildKafkaSpout("mytopic"), 1);
    builder.setBolt("print1", new PrintBolt(),1).shuffleGrouping("kafka");
}

澄清一下，在我重新启动管道之前一切正常。下面的行为是我可以在其他（非Storm）消费者身上得到的，也是我期望从KafkaSpout身上得到的

然而，Im使用默认设置获得的实际行为如下所示。消息被处理得很好，直到我停止管道，然后当我重新启动时，我得到了所有消息的重播，包括那些（a和B）我认为我已经处理过的消息

根据Matthias提到的配置选项，我可以将startoffsettime更改为latest，但这实际上是管道丢弃重新启动时产生的消息（消息“C”）的最新位置。

Answer 1

问题出现在提交代码中--如果Storm JAR在没有拓扑名称的情况下运行，并且本地集群没有捕获状态，因此没有重播，那么用于提交拓扑的模板代码将创建LocalCluster的实例。

所以

$ storm jar myjar.jar storm.myorg.MyTopology topologyname

我将在我的单节点开发集群上启动它，其中

$ storm jar myjar.jar storm.myorg.MyTopology