Apache Storm元组超时问题

Question

我试图在元组处理超时的情况下理解拓扑的状态（不是在三叉戟模式下）让我们假设在处理某个螺栓中的元组期间达到了超时阈值。在这种情况下，spout再次发出初始元组（具有与我理解的相同的消息ID）。现在让我们说Bolt完成处理元组并发出元组。在那种情况下：

1. 即使spout发出新的一个初始元组，失败的元组是否仍会继续由拓扑处理？
2. 如果是这样，acker的元组DAG将如何显示（因为有一个新的DAG创建了相同的初始元组id）前一个原始DAG会发生什么？
3. 当acker收到ack并发出前一个DAG的锚ID时会发生什么？

Answer 1

1：是的，失败的元组继续。这样做的原因是，试图阻止失败的元组继续进行是太昂贵了，因为喷口需要告诉所有关于失败的螺栓。

2：我认为这里有一个小小的误解。当spout发出元组时，消息id不是Storm用来在内部跟踪该元组DAG /树的信息。相反，spout执行器生成一个随机id（称之为rootId），并在本地存储rootId -> messageId的映射。消息ID永远不会离开spout执行器，也不会传播到螺栓。

当spout执行器向前发送元组时，它包括rootId。 rootId是用于识别元组树的acker和bolt所使用的。

最后，当树完全被激活或者元组失败时，spout执行器被告知相关的rootId成功或失败，并且它在其本地映射中查找原始的messageId。

由于具有相同messageId的新发射获得新的rootId，故障元组和新元组之间没有关系。 Storm被认为完全独立。

为了清楚起见，我简化了上面的内容，为了处理向多个螺栓发射的喷口，还有另一组随机ID（anchorId）。从概念上讲，你可以考虑你所拥有的情况

spout -> bolt1
      -> bolt2

就像拓扑一样处理

spout -> splitterBolt -> bolt1
                      -> bolt2

3：假设你的元组已经超时了。喷嘴执行人被告知rootId失败了。当发生这种情况时，spout执行程序调用spout.fail(msgId)，然后删除rootId -> messageId映射中的映射。

当acker收到ack时，如果树完全被激活，它可能会将ack发送到spout。当喷口接收到ack时，它与存储的rootId没有任何匹配，因此忽略了ack。

如果您有兴趣查看代码，可以在https://github.com/apache/storm/blob/b48e10559b65e834884d59887b30fc86d2988c20/storm-client/src/jvm/org/apache/storm/executor/spout/SpoutOutputCollectorImpl.java#L109找到它。 rootId -> messageId映射称为pending。