Storm拓扑中的可选流(Optional stream in Storm topology)

 我们有一个相当简单的风暴拓扑结构，只有一个头痛。 
 
 我们的一个螺栓可以找到它正在处理的数据是有效的，并且每个东西都正常地在流中继续，或者它可以发现它是无效但可修复的。 在这种情况下，我们需要发送它进行一些额外的处理。 
 
 我们尝试使用单独的bolt和stream将此步骤作为拓扑的一部分。 
 
declarer.declareStream(NORMAL_STREAM, getStreamFields());
declarer.declareStream(ERROR_STREAM, getErrorStreamFields());
 
 在execute方法的最后跟着一些类似下面的东西。 
 
if(errorOutput != null) {
    collector.emit(ERROR_STREAM, input, errorOutput);
}
else {
    collector.emit(NORMAL_STREAM, input, output);
}

collector.ack(input);
 
 这确实有效但是它具有破坏性的效果，导致所有不沿着这个错误路径的元组失败并且无休止地重新发送喷口。 
 
 我认为这是因为错误螺栓不能发送它没有收到的消息的acks但是acker的东西在将ack发送回spout之前等待拓扑中的所有螺栓。 至少取出错误处理螺栓会导致所有东西正确地回到喷口。 
 
 实现这样的事情的最佳方法是什么？ 

We have a fairly simple storm topology with one head ache.
 
One of our bolts can either find the data it is processing to be valid and every thing carries on down the stream as normal or it can find it to be invalid but fixable. In which case we need to send it for some additional processing.
 
We tried making this step part of the topology with a separate bolt and stream.
 
declarer.declareStream(NORMAL_STREAM, getStreamFields());
declarer.declareStream(ERROR_STREAM, getErrorStreamFields());
 
Followed by some thing like the following at the end of the execute method.
 
if(errorOutput != null) {
    collector.emit(ERROR_STREAM, input, errorOutput);
}
else {
    collector.emit(NORMAL_STREAM, input, output);
}

collector.ack(input);
 
This does work however it has a breaking effect of causing all of the tuples that do not go down this error path to fail and get re-sent by the spout endlessly.
 
I think this is because the error bolt can not send acks for messages it doesn't receive but the acker thing waits for all the bolts in a topology to ack before sending the ack back to the spout. At the very least taking out the error processing bolt causes every thing to get acked back to the spout correctly.
 
What is the best way to achieve some thing like this?

原文：https://stackoverflow.com/questions/28047372