Streaming fault tolerance with event sending

[Hermann Gábor <re...@gmail.com>]

Hey,

We've been implementing a simple Storm-like fault tolerance system
with persisting source records and keeping track of all the records
(whether they've been processed), and replaying them if they fail. The
straightforward way to do this was creating a special AbstractJobVertex
(and AbstractInvokable) that deals with replaying and keeps track of the
records.

So there is a so called FTLayerVertex that needs a two-way
communication with the tasks:

   1. it needs to be notified about a processed record or a newly created
   record
   2. it needs to resend failed records to the appropriate tasks

We've decided to use events for 1. and regular message sending (i.e.
BufferReader/Writer) for 2. So we need to send events backward on a
BufferReader.

The problem is that backward events are only sent and received when
the channel is initialized (in case of one channel for now), there's a
living
connection (i.e. it's not UNKNOWN, please correct me if I'm wrong). But
sometimes the connection is not yet initialized when we're sending these
events from the tasks to the FTLayerVertex.

We've tried a workaround: sending an "initializer" forward event from FT
to all the tasks, and waiting for this event at the beginning of every
task's
invoke, thus the backward events will only be sent when the channel is
surely up. That did not work out, because in order to receive an event,
we need to read that input, and we only want to read and process that
input after we made sure we can use the channel.

Another workaround were sending an "initializer" record instead of an
event. That did not work out either, because we wanted to read that
input (from the FTLayerVertex) as a regular input, so the given
BufferReader were also passed to a UnionBufferReader, and reading
the given BufferReader separately (wrapped in a RecordReader)
resulted in the UnionBufferReader getting stuck. (It seemed as the
UnionBufferReader put the given BufferReader in it's queue because of
this "initializer" message, but the message had been already read from
the BufferReader separately, so it got stuck waiting for a new record
from the BufferReader).

These workarounds don't seem natural either, because we're pulling this
"synchronization" of connection initialization a layer up. It would be nice
if we had some kind of events that are made sure to get to the
destination. Is that possible? Would that be useful for other purposes too?

Does anyone know some workaround that might work? Are we missing
something? (Please correct me if I see something wrong.) Should we use
something else instead of events? (I think we cannot avoid this two-way
communication between vertices.) All suggestions are welcome!

Best regards,
Gabor