Правильный способ ACK в шторме в цепочке болтов

Просто хочу убедиться, что я получил, как Ack-ing работает в Storm. У меня есть 1 носик и 2 болта вместе. Spout испускает кортеж в Bolt1, который, в свою очередь, выдает кортеж в Bolt 2. Я хочу, чтобы Bolt 2 получил исходный кортеж, отправленный из Spout, и я не уверен, как это сделать.

Чтобы гарантировать отказоустойчивость (т.е.: кортежи возмущены), я хочу, чтобы в болт 2 кортеж, испускаемый Spout, на случай, если он не сработает где-то в процессе, чтобы он мог быть возмущен.

Рассмотрим следующий пример:

Spout:

 _collector.emit(new Values(queue.dequeue())

Болт1:

def execute(tuple: Tuple) {
 _collector.emit(tuple, new Values("stuff"))
}

В этот момент кортеж - это кортеж, посланный носиком. Я могу проверить это здесь без проблем. Теперь добавьте еще один болт, который прослушивается на кортежах, испускаемых Bolt1.

Болт2:

def execute(tuple2: Tuple) {
 _collector.emit(tuple2, new Values("foo"))
}

В этот момент кортеж в tuple2 является кортежем, посланным из Bolt1 (тем, у которого в нем есть строка "stuff" ).
Поэтому, если я отправлю ack в Bolt2, это будет привязывать кортеж от Bolt1, а не тот, который отправлен из Spout. Верный?

Как я могу получить кортеж, который был отправлен из носика? Должен ли я вернуть исходный носик на все остальные носики, чтобы я мог получить его в последнем Болте и сказать?

Я читал учебники Nathan, и у меня создалось впечатление, что я могу получить кортеж, полученный в Bolt1 (от Spout) прямо после того, как исправив tuple2. Это свяжет недавно испущенный кортеж2 с исходным кортежем, посланным Spout, поэтому, когда Bolt2 acks tuple 2, он фактически берет исходный кортеж из Spout. Это правда?

Дайте мне знать, если я пропущу что-то в своем объяснении.
Большое спасибо!

Ответ 1

Для тех, кого это интересует, я нашел решение, попросив группу бури. Мне нужно в Spout, чтобы испускать кортежи следующим образом (с уникальным идентификатором):

Spout:

 //ties in tuple to this UID
 _collector.emit(new Values(queue.dequeue(), *uniqueID*)

Затем Bolt1 будет кормить кортеж только после того, как он испустит его в Bolt2

Болт1:

 //emit first then ack
 _collector.emit(tuple, new Values("stuff")) //**anchoring** - read below to see what it means
 _collector.ack(tuple)

В этот момент кортеж из Spout был выделен в Bolt1, но в то же время недавно выпущенный "материал" для Bolt2 "привязан" к кортежу от Spout. Это означает, что он по-прежнему должен быть отброшен позже, иначе при тайм-ауте он будет возмущен носиком.

Bolt2:

 _collector.ack(tuple)

Bolt2 должен получить кортеж, полученный от Bolt1, который отправит в последний раз, который ожидал Spout. Если в этот момент Bolt2 испускает кортеж, тогда должен быть Bolt3, который получит его и получит его. Если кортеж не будет отмечен в последней точке, Spout запустит его и повторно отправит.

Каждый раз, когда привязка выполняется в инструкции emit от болта к болту, создается новый node в структуре "tree"... ну, скорее, как список в моем случае, так как я никогда не отправляю один и тот же кортеж до 2 или более кортежей, у меня есть отношения 1 к 1.

Все узлы в дереве должны быть выделены, и только тогда кортеж будет отмечен как полностью прибывший. Если кортеж не отключен, и он отправляется с UID и позже привязывается к нему, он будет храниться в памяти навсегда (до тех пор, пока не будет найден).

Надеюсь, это поможет.

Ответ 2

Вам нужно anchor кортеж. Взгляните на https://github.com/nathanmarz/storm/wiki/Guaranteeing-message-processing И особенно вам это нужно:

List<Tuple> anchors = new ArrayList<Tuple>();
anchors.add(tuple1);
anchors.add(tuple2);
_collector.emit(anchors, new Values(1, 2, 3));

Ответ 3

Об этом можно прочитать в официальной документации .

Если вы хотите отслеживать выполнение своих задач по всем болтам, вы можете использовать BaseBasicBolt в качестве родительского класса, где это поведение уже определено.

В любом другом случае использования (т.е. вы хотите, чтобы кортеж был вставлен перед исполнением в последнем болте), вы должны вручную определить связи между кортежами (называемыми привязкой). См. Документацию.