为什么Trident在这个最小示例中不调用ack（）或fail（）？

Question

我试图在三叉戟中创建一个小例子。目标是查看在失败的情况下如何重播元组。下面是拓扑定义

        Random rand = new Random();

        Config config = new Config();
        config.setDebug(true);
        config.setNumWorkers(1);

        TridentTopology topology = new TridentTopology();

        topology.newStream("spout", new RandomIntegerSpout())
                .map((MapFunction) tridentTuple -> {
                    if ((tridentTuple.getLongByField("msgid") % 50 == 0) &&
                            (rand.nextInt(2) == 1)) {
                        System.out.println(String.format("Failed to process tuple %d", tridentTuple.getLongByField("msgid")));
                        throw new ReportedFailedException("Divisible by 50");
                    }
                    return new Values(tridentTuple.toArray());
                })
                .peek((Consumer) tridentTuple -> System.out.println(tridentTuple.getValues()));

我使用了来自storm-starter的RandomIntegerSpout，它扩展了BaseRichSpout，并且只生成随机数。然后应用mapfunction，它只是每50个元组抽取一个随机数，并随机使该元组失败。

问题是，我没有得到任何acks或fails。

通过检查，看到代码几乎是微不足道的，我可能误解了关于Trident/Storm的一些基本内容。如果批处理完成，我期望trident调用spout的和ack方法是错误的吗？我意识到在ibatchspout中没有fail方法。Trident如何处理批次的重放？？

Answer 1

Trident spouts不会在单个元组级别上ack或失败元组。相反，元组被作为一个批处理来处理。

三叉戟喷口通常看起来像这样的接口。

M emitPartitionBatch(TransactionAttempt tx, TridentCollector collector, PartitionT partition, M lastPartitionMeta);

其思想是，Trident将管理对批处理元组的ACK/Failles的跟踪，然后如果批处理失败，它将请求spout重复该批处理，如果不重复，它就不重复。

我认为您在Storm UI中没有看到任何与此相关的内容的原因是irichbolt实际上并没有被表示出来。相反，它被包装了，因此ack/fail调用是在spout-spout组件的“引擎盖下”发生的。如果您想确定是否正在调用ACK/FAIL，请尝试向irichspout的ACK/FAIL方法添加一些日志记录。